Skip to content
This repository was archived by the owner on May 8, 2026. It is now read-only.

Commit feaec40

Browse files
author
Teng Zhong
committed
Add ListChangeStreamPartitions callable
1 parent 6a4444f commit feaec40

6 files changed

Lines changed: 458 additions & 0 deletions

File tree

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/BigtableDataClient.java‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import com.google.cloud.bigtable.data.v2.models.Filters.Filter;
3636
import com.google.cloud.bigtable.data.v2.models.KeyOffset;
3737
import com.google.cloud.bigtable.data.v2.models.Query;
38+
import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange;
3839
import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
3940
import com.google.cloud.bigtable.data.v2.models.Row;
4041
import com.google.cloud.bigtable.data.v2.models.RowAdapter;
@@ -1489,6 +1490,141 @@ public UnaryCallable<ReadModifyWriteRow, Row> readModifyWriteRowCallable() {
14891490
return stub.readModifyWriteRowCallable();
14901491
}
14911492

1493+
/**
1494+
* Convenience method for synchronously streaming the partitions of a table. The returned
1495+
* ServerStream instance is not threadsafe, it can only be used from single thread.
1496+
*
1497+
* <p>Sample code:
1498+
*
1499+
* <pre>{@code
1500+
* try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
1501+
* String tableId = "[TABLE]";
1502+
*
1503+
* try {
1504+
* ServerStream<ByteStringRange> stream = bigtableDataClient.listChangeStreamPartitions(tableId);
1505+
* int count = 0;
1506+
*
1507+
* // Iterator style
1508+
* for (ByteStringRange partition : stream) {
1509+
* if (++count > 10) {
1510+
* stream.cancel();
1511+
* break;
1512+
* }
1513+
* // Do something with partition
1514+
* }
1515+
* } catch (NotFoundException e) {
1516+
* System.out.println("Tried to read a non-existent table");
1517+
* } catch (RuntimeException e) {
1518+
* e.printStackTrace();
1519+
* }
1520+
* }
1521+
* }</pre>
1522+
*
1523+
* @see ServerStreamingCallable For call styles.
1524+
*/
1525+
public ServerStream<ByteStringRange> listChangeStreamPartitions(String tableId) {
1526+
return listChangeStreamPartitionsCallable().call(tableId);
1527+
}
1528+
1529+
/**
1530+
* Convenience method for asynchronously streaming the partitions of a table.
1531+
*
1532+
* <p>Sample code:
1533+
*
1534+
* <pre>{@code
1535+
* try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
1536+
* String tableId = "[TABLE]";
1537+
*
1538+
* bigtableDataClient.listChangeStreamPartitionsAsync(tableId, new ResponseObserver<ByteStringRange>() {
1539+
* StreamController controller;
1540+
* int count = 0;
1541+
*
1542+
* public void onStart(StreamController controller) {
1543+
* this.controller = controller;
1544+
* }
1545+
* public void onResponse(ByteStringRange partition) {
1546+
* if (++count > 10) {
1547+
* controller.cancel();
1548+
* return;
1549+
* }
1550+
* // Do something with partition
1551+
* }
1552+
* public void onError(Throwable t) {
1553+
* if (t instanceof NotFoundException) {
1554+
* System.out.println("Tried to read a non-existent table");
1555+
* } else {
1556+
* t.printStackTrace();
1557+
* }
1558+
* }
1559+
* public void onComplete() {
1560+
* // Handle stream completion
1561+
* }
1562+
* });
1563+
* }
1564+
* }</pre>
1565+
*/
1566+
public void listChangeStreamPartitionsAsync(
1567+
String tableId, ResponseObserver<ByteStringRange> observer) {
1568+
listChangeStreamPartitionsCallable().call(tableId, observer);
1569+
}
1570+
1571+
/**
1572+
* Streams back the results of the query. The returned callable object allows for customization of
1573+
* api invocation.
1574+
*
1575+
* <p>Sample code:
1576+
*
1577+
* <pre>{@code
1578+
* try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
1579+
* String tableId = "[TABLE]";
1580+
*
1581+
* // Iterator style
1582+
* try {
1583+
* for(ByteStringRange partition : bigtableDataClient.listChangeStreamPartitionsCallable().call(tableId)) {
1584+
* // Do something with partition
1585+
* }
1586+
* } catch (NotFoundException e) {
1587+
* System.out.println("Tried to read a non-existent table");
1588+
* } catch (RuntimeException e) {
1589+
* e.printStackTrace();
1590+
* }
1591+
*
1592+
* // Sync style
1593+
* try {
1594+
* List<ByteStringRange> partitions = bigtableDataClient.listChangeStreamPartitionsCallable().all().call(tableId);
1595+
* } catch (NotFoundException e) {
1596+
* System.out.println("Tried to read a non-existent table");
1597+
* } catch (RuntimeException e) {
1598+
* e.printStackTrace();
1599+
* }
1600+
*
1601+
* // Point look up
1602+
* ApiFuture<ByteStringRange> partitionFuture =
1603+
* bigtableDataClient.listChangeStreamPartitionsCallable().first().futureCall(tableId);
1604+
*
1605+
* ApiFutures.addCallback(partitionFuture, new ApiFutureCallback<ByteStringRange>() {
1606+
* public void onFailure(Throwable t) {
1607+
* if (t instanceof NotFoundException) {
1608+
* System.out.println("Tried to read a non-existent table");
1609+
* } else {
1610+
* t.printStackTrace();
1611+
* }
1612+
* }
1613+
* public void onSuccess(ByteStringRange result) {
1614+
* System.out.println("Got partition: " + result);
1615+
* }
1616+
* }, MoreExecutors.directExecutor());
1617+
*
1618+
* // etc
1619+
* }
1620+
* }</pre>
1621+
*
1622+
* @see ServerStreamingCallable For call styles.
1623+
*/
1624+
public ServerStreamingCallable<String, ByteStringRange> listChangeStreamPartitionsCallable() {
1625+
return stub.listChangeStreamPartitionsCallable();
1626+
}
1627+
14921628
/** Close the clients and releases all associated resources. */
14931629
@Override
14941630
public void close() {

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStub.java‎

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,8 @@
4747
import com.google.bigtable.v2.BigtableGrpc;
4848
import com.google.bigtable.v2.CheckAndMutateRowRequest;
4949
import com.google.bigtable.v2.CheckAndMutateRowResponse;
50+
import com.google.bigtable.v2.ListChangeStreamPartitionsRequest;
51+
import com.google.bigtable.v2.ListChangeStreamPartitionsResponse;
5052
import com.google.bigtable.v2.MutateRowRequest;
5153
import com.google.bigtable.v2.MutateRowResponse;
5254
import com.google.bigtable.v2.MutateRowsRequest;
@@ -65,11 +67,13 @@
6567
import com.google.cloud.bigtable.data.v2.models.DefaultRowAdapter;
6668
import com.google.cloud.bigtable.data.v2.models.KeyOffset;
6769
import com.google.cloud.bigtable.data.v2.models.Query;
70+
import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange;
6871
import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
6972
import com.google.cloud.bigtable.data.v2.models.Row;
7073
import com.google.cloud.bigtable.data.v2.models.RowAdapter;
7174
import com.google.cloud.bigtable.data.v2.models.RowMutation;
7275
import com.google.cloud.bigtable.data.v2.models.RowMutationEntry;
76+
import com.google.cloud.bigtable.data.v2.stub.changestream.ListChangeStreamPartitionsUserCallable;
7377
import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracerStreamingCallable;
7478
import com.google.cloud.bigtable.data.v2.stub.metrics.BigtableTracerUnaryCallable;
7579
import com.google.cloud.bigtable.data.v2.stub.metrics.BuiltinMetricsTracerFactory;
@@ -142,6 +146,8 @@ public class EnhancedBigtableStub implements AutoCloseable {
142146
private final UnaryCallable<ConditionalRowMutation, Boolean> checkAndMutateRowCallable;
143147
private final UnaryCallable<ReadModifyWriteRow, Row> readModifyWriteRowCallable;
144148

149+
private final ServerStreamingCallable<String, ByteStringRange> listChangeStreamPartitionsCallable;
150+
145151
public static EnhancedBigtableStub create(EnhancedBigtableStubSettings settings)
146152
throws IOException {
147153
settings = finalizeSettings(settings, Tags.getTagger(), Stats.getStatsRecorder());
@@ -284,6 +290,7 @@ public EnhancedBigtableStub(EnhancedBigtableStubSettings settings, ClientContext
284290
bulkMutateRowsCallable = createBulkMutateRowsCallable();
285291
checkAndMutateRowCallable = createCheckAndMutateRowCallable();
286292
readModifyWriteRowCallable = createReadModifyWriteRowCallable();
293+
listChangeStreamPartitionsCallable = createListChangeStreamPartitionsCallable();
287294
}
288295

289296
// <editor-fold desc="Callable creators">
@@ -798,6 +805,71 @@ public Map<String, String> extract(ReadModifyWriteRowRequest request) {
798805
methodName, new ReadModifyWriteRowCallable(retrying, requestContext));
799806
}
800807

808+
/**
809+
* Creates a callable chain to handle streaming ListChangeStreamPartitions RPCs. The chain will:
810+
*
811+
* <ul>
812+
* <li>Convert a String format tableId into a {@link com.google.bigtable.v2.ReadChangeStreamRequest} and
813+
* dispatch the RPC.
814+
* <li>Upon receiving the response stream, it will convert the {@link
815+
* com.google.bigtable.v2.ListChangeStreamPartitionsResponse}s into {@link ByteStringRange}.
816+
* </ul>
817+
*/
818+
private ServerStreamingCallable<String, ByteStringRange>
819+
createListChangeStreamPartitionsCallable() {
820+
ServerStreamingCallable<ListChangeStreamPartitionsRequest, ListChangeStreamPartitionsResponse>
821+
base =
822+
GrpcRawCallableFactory.createServerStreamingCallable(
823+
GrpcCallSettings
824+
.<ListChangeStreamPartitionsRequest, ListChangeStreamPartitionsResponse>
825+
newBuilder()
826+
.setMethodDescriptor(BigtableGrpc.getListChangeStreamPartitionsMethod())
827+
.setParamsExtractor(
828+
new RequestParamsExtractor<ListChangeStreamPartitionsRequest>() {
829+
@Override
830+
public Map<String, String> extract(
831+
ListChangeStreamPartitionsRequest listChangeStreamPartitionsRequest) {
832+
return ImmutableMap.of(
833+
"table_name",
834+
listChangeStreamPartitionsRequest.getTableName(),
835+
"app_profile_id",
836+
listChangeStreamPartitionsRequest.getAppProfileId());
837+
}
838+
})
839+
.build(),
840+
settings.listChangeStreamPartitionsSettings().getRetryableCodes());
841+
842+
ServerStreamingCallable<String, ByteStringRange> userCallable =
843+
new ListChangeStreamPartitionsUserCallable(base, requestContext);
844+
845+
ServerStreamingCallable<String, ByteStringRange> withStatsHeaders =
846+
new StatsHeadersServerStreamingCallable<>(userCallable);
847+
848+
// Copy settings for the middle String -> ByteStringRange callable (as opposed to the inner
849+
// ListChangeStreamPartitionsRequest -> ListChangeStreamPartitionsResponse callable).
850+
ServerStreamingCallSettings<String, ByteStringRange> innerSettings =
851+
ServerStreamingCallSettings.<String, ByteStringRange>newBuilder()
852+
.setRetryableCodes(settings.listChangeStreamPartitionsSettings().getRetryableCodes())
853+
.setRetrySettings(settings.listChangeStreamPartitionsSettings().getRetrySettings())
854+
.setIdleTimeout(settings.listChangeStreamPartitionsSettings().getIdleTimeout())
855+
.build();
856+
857+
ServerStreamingCallable<String, ByteStringRange> watched =
858+
Callables.watched(withStatsHeaders, innerSettings, clientContext);
859+
860+
ServerStreamingCallable<String, ByteStringRange> withBigtableTracer =
861+
new BigtableTracerStreamingCallable<>(watched);
862+
863+
ServerStreamingCallable<String, ByteStringRange> retrying =
864+
Callables.retrying(withBigtableTracer, innerSettings, clientContext);
865+
866+
SpanName span = getSpanName("ListChangeStreamPartitions");
867+
ServerStreamingCallable<String, ByteStringRange> traced =
868+
new TracedServerStreamingCallable<>(retrying, clientContext.getTracerFactory(), span);
869+
870+
return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
871+
}
872+
801873
/**
802874
* Wraps a callable chain in a user presentable callable that will inject the default call context
803875
* and trace the call.
@@ -854,6 +926,11 @@ public UnaryCallable<ConditionalRowMutation, Boolean> checkAndMutateRowCallable(
854926
public UnaryCallable<ReadModifyWriteRow, Row> readModifyWriteRowCallable() {
855927
return readModifyWriteRowCallable;
856928
}
929+
930+
/** Returns a streaming list change stream partitions callable */
931+
public ServerStreamingCallable<String, ByteStringRange> listChangeStreamPartitionsCallable() {
932+
return listChangeStreamPartitionsCallable;
933+
}
857934
// </editor-fold>
858935

859936
private SpanName getSpanName(String methodName) {

‎google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettings.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
3838
import com.google.cloud.bigtable.data.v2.models.KeyOffset;
3939
import com.google.cloud.bigtable.data.v2.models.Query;
40+
import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange;
4041
import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
4142
import com.google.cloud.bigtable.data.v2.models.Row;
4243
import com.google.cloud.bigtable.data.v2.models.RowMutation;
@@ -137,6 +138,22 @@ public class EnhancedBigtableStubSettings extends StubSettings<EnhancedBigtableS
137138
.setTotalTimeout(Duration.ofMinutes(10))
138139
.build();
139140

141+
private static final Set<Code> LIST_CHANGE_STREAM_PARTITIONS_RETRY_CODES =
142+
ImmutableSet.<Code>builder().addAll(IDEMPOTENT_RETRY_CODES).add(Code.ABORTED).build();
143+
144+
private static final RetrySettings LIST_CHANGE_STREAM_PARTITIONS_RETRY_SETTINGS =
145+
RetrySettings.newBuilder()
146+
.setInitialRetryDelay(Duration.ofMillis(10))
147+
.setRetryDelayMultiplier(2.0)
148+
.setMaxRetryDelay(Duration.ofMinutes(1))
149+
.setMaxAttempts(10)
150+
.setJittered(true)
151+
.setInitialRpcTimeout(Duration.ofMinutes(5))
152+
.setRpcTimeoutMultiplier(2.0)
153+
.setMaxRpcTimeout(Duration.ofMinutes(5))
154+
.setTotalTimeout(Duration.ofHours(12))
155+
.build();
156+
140157
/**
141158
* Scopes that are equivalent to JWT's audience.
142159
*
@@ -174,6 +191,9 @@ public class EnhancedBigtableStubSettings extends StubSettings<EnhancedBigtableS
174191
private final UnaryCallSettings<ConditionalRowMutation, Boolean> checkAndMutateRowSettings;
175192
private final UnaryCallSettings<ReadModifyWriteRow, Row> readModifyWriteRowSettings;
176193

194+
private final ServerStreamingCallSettings<String, ByteStringRange>
195+
listChangeStreamPartitionsSettings;
196+
177197
private EnhancedBigtableStubSettings(Builder builder) {
178198
super(builder);
179199

@@ -208,6 +228,7 @@ private EnhancedBigtableStubSettings(Builder builder) {
208228
bulkReadRowsSettings = builder.bulkReadRowsSettings.build();
209229
checkAndMutateRowSettings = builder.checkAndMutateRowSettings.build();
210230
readModifyWriteRowSettings = builder.readModifyWriteRowSettings.build();
231+
listChangeStreamPartitionsSettings = builder.listChangeStreamPartitionsSettings.build();
211232
}
212233

213234
/** Create a new builder. */
@@ -491,6 +512,10 @@ public UnaryCallSettings<ReadModifyWriteRow, Row> readModifyWriteRowSettings() {
491512
return readModifyWriteRowSettings;
492513
}
493514

515+
public ServerStreamingCallSettings<String, ByteStringRange> listChangeStreamPartitionsSettings() {
516+
return listChangeStreamPartitionsSettings;
517+
}
518+
494519
/** Returns a builder containing all the values of this settings class. */
495520
public Builder toBuilder() {
496521
return new Builder(this);
@@ -516,6 +541,9 @@ public static class Builder extends StubSettings.Builder<EnhancedBigtableStubSet
516541
checkAndMutateRowSettings;
517542
private final UnaryCallSettings.Builder<ReadModifyWriteRow, Row> readModifyWriteRowSettings;
518543

544+
private final ServerStreamingCallSettings.Builder<String, ByteStringRange>
545+
listChangeStreamPartitionsSettings;
546+
519547
/**
520548
* Initializes a new Builder with sane defaults for all settings.
521549
*
@@ -626,6 +654,12 @@ private Builder() {
626654

627655
readModifyWriteRowSettings = UnaryCallSettings.newUnaryCallSettingsBuilder();
628656
copyRetrySettings(baseDefaults.readModifyWriteRowSettings(), readModifyWriteRowSettings);
657+
658+
listChangeStreamPartitionsSettings = ServerStreamingCallSettings.newBuilder();
659+
listChangeStreamPartitionsSettings
660+
.setRetryableCodes(LIST_CHANGE_STREAM_PARTITIONS_RETRY_CODES)
661+
.setRetrySettings(LIST_CHANGE_STREAM_PARTITIONS_RETRY_SETTINGS)
662+
.setIdleTimeout(Duration.ofMinutes(5));
629663
}
630664

631665
private Builder(EnhancedBigtableStubSettings settings) {
@@ -646,6 +680,7 @@ private Builder(EnhancedBigtableStubSettings settings) {
646680
bulkReadRowsSettings = settings.bulkReadRowsSettings.toBuilder();
647681
checkAndMutateRowSettings = settings.checkAndMutateRowSettings.toBuilder();
648682
readModifyWriteRowSettings = settings.readModifyWriteRowSettings.toBuilder();
683+
listChangeStreamPartitionsSettings = settings.listChangeStreamPartitionsSettings.toBuilder();
649684
}
650685
// <editor-fold desc="Private Helpers">
651686

@@ -857,6 +892,7 @@ public String toString() {
857892
.add("bulkReadRowsSettings", bulkReadRowsSettings)
858893
.add("checkAndMutateRowSettings", checkAndMutateRowSettings)
859894
.add("readModifyWriteRowSettings", readModifyWriteRowSettings)
895+
.add("listChangeStreamPartitionsSettings", listChangeStreamPartitionsSettings)
860896
.add("parent", super.toString())
861897
.toString();
862898
}

0 commit comments

Comments
 (0)