Sample code: + * + *
{@code
+ * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+ * String tableId = "[TABLE]";
+ *
+ * try {
+ * ServerStream stream = bigtableDataClient.listChangeStreamPartitions(tableId);
+ * int count = 0;
+ *
+ * // Iterator style
+ * for (RowRange partition : stream) {
+ * if (++count > 10) {
+ * stream.cancel();
+ * break;
+ * }
+ * // Do something with partition
+ * }
+ * } catch (NotFoundException e) {
+ * System.out.println("Tried to read a non-existent table");
+ * } catch (RuntimeException e) {
+ * e.printStackTrace();
+ * }
+ * }
+ * }
+ *
+ * @see ServerStreamingCallable For call styles.
+ */
+ @InternalApi("Used in Changestream beam pipeline.")
+ public ServerStreamSample code: + * + *
{@code
+ * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+ * String tableId = "[TABLE]";
+ *
+ * bigtableDataClient.listChangeStreamPartitionsAsync(tableId, new ResponseObserver() {
+ * StreamController controller;
+ * int count = 0;
+ *
+ * public void onStart(StreamController controller) {
+ * this.controller = controller;
+ * }
+ * public void onResponse(RowRange partition) {
+ * if (++count > 10) {
+ * controller.cancel();
+ * return;
+ * }
+ * // Do something with partition
+ * }
+ * public void onError(Throwable t) {
+ * if (t instanceof NotFoundException) {
+ * System.out.println("Tried to read a non-existent table");
+ * } else {
+ * t.printStackTrace();
+ * }
+ * }
+ * public void onComplete() {
+ * // Handle stream completion
+ * }
+ * });
+ * }
+ * }
+ */
+ @InternalApi("Used in Changestream beam pipeline.")
+ public void listChangeStreamPartitionsAsync(String tableId, ResponseObserverSample code: + * + *
{@code
+ * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+ * String tableId = "[TABLE]";
+ *
+ * // Iterator style
+ * try {
+ * for(RowRange partition : bigtableDataClient.listChangeStreamPartitionsCallable().call(tableId)) {
+ * // Do something with partition
+ * }
+ * } catch (NotFoundException e) {
+ * System.out.println("Tried to read a non-existent table");
+ * } catch (RuntimeException e) {
+ * e.printStackTrace();
+ * }
+ *
+ * // Sync style
+ * try {
+ * List partitions = bigtableDataClient.listChangeStreamPartitionsCallable().all().call(tableId);
+ * } catch (NotFoundException e) {
+ * System.out.println("Tried to read a non-existent table");
+ * } catch (RuntimeException e) {
+ * e.printStackTrace();
+ * }
+ *
+ * // Point look up
+ * ApiFuture partitionFuture =
+ * bigtableDataClient.listChangeStreamPartitionsCallable().first().futureCall(tableId);
+ *
+ * ApiFutures.addCallback(partitionFuture, new ApiFutureCallback() {
+ * public void onFailure(Throwable t) {
+ * if (t instanceof NotFoundException) {
+ * System.out.println("Tried to read a non-existent table");
+ * } else {
+ * t.printStackTrace();
+ * }
+ * }
+ * public void onSuccess(RowRange result) {
+ * System.out.println("Got partition: " + result);
+ * }
+ * }, MoreExecutors.directExecutor());
+ *
+ * // etc
+ * }
+ * }
+ *
+ * @see ServerStreamingCallable For call styles.
+ */
+ @InternalApi("Used in Changestream beam pipeline.")
+ public ServerStreamingCallablebuilder().addAll(IDEMPOTENT_RETRY_CODES).add(Code.ABORTED).build();
+
+ private static final RetrySettings LIST_CHANGE_STREAM_PARTITIONS_RETRY_SETTINGS =
+ RetrySettings.newBuilder()
+ .setInitialRetryDelay(Duration.ofMillis(10))
+ .setRetryDelayMultiplier(2.0)
+ .setMaxRetryDelay(Duration.ofMinutes(1))
+ .setMaxAttempts(10)
+ .setJittered(true)
+ .setInitialRpcTimeout(Duration.ofMinutes(1))
+ .setRpcTimeoutMultiplier(2.0)
+ .setMaxRpcTimeout(Duration.ofMinutes(10))
+ .setTotalTimeout(Duration.ofMinutes(60))
+ .build();
+
/**
* Scopes that are equivalent to JWT's audience.
*
@@ -174,6 +191,8 @@ public class EnhancedBigtableStubSettings extends StubSettings checkAndMutateRowSettings;
private final UnaryCallSettings readModifyWriteRowSettings;
+ private final ServerStreamingCallSettings listChangeStreamPartitionsSettings;
+
private EnhancedBigtableStubSettings(Builder builder) {
super(builder);
@@ -208,6 +227,7 @@ private EnhancedBigtableStubSettings(Builder builder) {
bulkReadRowsSettings = builder.bulkReadRowsSettings.build();
checkAndMutateRowSettings = builder.checkAndMutateRowSettings.build();
readModifyWriteRowSettings = builder.readModifyWriteRowSettings.build();
+ listChangeStreamPartitionsSettings = builder.listChangeStreamPartitionsSettings.build();
}
/** Create a new builder. */
@@ -491,6 +511,10 @@ public UnaryCallSettings readModifyWriteRowSettings() {
return readModifyWriteRowSettings;
}
+ public ServerStreamingCallSettings listChangeStreamPartitionsSettings() {
+ return listChangeStreamPartitionsSettings;
+ }
+
/** Returns a builder containing all the values of this settings class. */
public Builder toBuilder() {
return new Builder(this);
@@ -516,6 +540,9 @@ public static class Builder extends StubSettings.Builder readModifyWriteRowSettings;
+ private final ServerStreamingCallSettings.Builder
+ listChangeStreamPartitionsSettings;
+
/**
* Initializes a new Builder with sane defaults for all settings.
*
@@ -626,6 +653,12 @@ private Builder() {
readModifyWriteRowSettings = UnaryCallSettings.newUnaryCallSettingsBuilder();
copyRetrySettings(baseDefaults.readModifyWriteRowSettings(), readModifyWriteRowSettings);
+
+ listChangeStreamPartitionsSettings = ServerStreamingCallSettings.newBuilder();
+ listChangeStreamPartitionsSettings
+ .setRetryableCodes(LIST_CHANGE_STREAM_PARTITIONS_RETRY_CODES)
+ .setRetrySettings(LIST_CHANGE_STREAM_PARTITIONS_RETRY_SETTINGS)
+ .setIdleTimeout(Duration.ofMinutes(5));
}
private Builder(EnhancedBigtableStubSettings settings) {
@@ -646,6 +679,7 @@ private Builder(EnhancedBigtableStubSettings settings) {
bulkReadRowsSettings = settings.bulkReadRowsSettings.toBuilder();
checkAndMutateRowSettings = settings.checkAndMutateRowSettings.toBuilder();
readModifyWriteRowSettings = settings.readModifyWriteRowSettings.toBuilder();
+ listChangeStreamPartitionsSettings = settings.listChangeStreamPartitionsSettings.toBuilder();
}
//
@@ -857,6 +891,7 @@ public String toString() {
.add("bulkReadRowsSettings", bulkReadRowsSettings)
.add("checkAndMutateRowSettings", checkAndMutateRowSettings)
.add("readModifyWriteRowSettings", readModifyWriteRowSettings)
+ .add("listChangeStreamPartitionsSettings", listChangeStreamPartitionsSettings)
.add("parent", super.toString())
.toString();
}
diff --git a/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/changestream/ListChangeStreamPartitionsUserCallable.java b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/changestream/ListChangeStreamPartitionsUserCallable.java
new file mode 100644
index 000000000000..1d3393bb2b9f
--- /dev/null
+++ b/google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/changestream/ListChangeStreamPartitionsUserCallable.java
@@ -0,0 +1,93 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.stub.changestream;
+
+import com.google.api.gax.rpc.ApiCallContext;
+import com.google.api.gax.rpc.ResponseObserver;
+import com.google.api.gax.rpc.ServerStreamingCallable;
+import com.google.api.gax.rpc.StreamController;
+import com.google.bigtable.v2.ListChangeStreamPartitionsRequest;
+import com.google.bigtable.v2.ListChangeStreamPartitionsResponse;
+import com.google.bigtable.v2.RowRange;
+import com.google.cloud.bigtable.data.v2.internal.NameUtil;
+import com.google.cloud.bigtable.data.v2.internal.RequestContext;
+
+/** Simple wrapper for ListChangeStreamPartitions to wrap the request and response protobufs. */
+public class ListChangeStreamPartitionsUserCallable
+ extends ServerStreamingCallable {
+ private final RequestContext requestContext;
+ private final ServerStreamingCallable<
+ ListChangeStreamPartitionsRequest, ListChangeStreamPartitionsResponse>
+ inner;
+
+ public ListChangeStreamPartitionsUserCallable(
+ ServerStreamingCallable
+ inner,
+ RequestContext requestContext) {
+ this.requestContext = requestContext;
+ this.inner = inner;
+ }
+
+ @Override
+ public void call(
+ String tableId, ResponseObserver responseObserver, ApiCallContext context) {
+ String tableName =
+ NameUtil.formatTableName(
+ requestContext.getProjectId(), requestContext.getInstanceId(), tableId);
+ ListChangeStreamPartitionsRequest request =
+ ListChangeStreamPartitionsRequest.newBuilder()
+ .setTableName(tableName)
+ .setAppProfileId(requestContext.getAppProfileId())
+ .build();
+
+ inner.call(request, new ConvertPartitionToRangeObserver(responseObserver), context);
+ }
+
+ private class ConvertPartitionToRangeObserver
+ implements ResponseObserver {
+
+ private final ResponseObserver outerObserver;
+
+ ConvertPartitionToRangeObserver(ResponseObserver observer) {
+ this.outerObserver = observer;
+ }
+
+ @Override
+ public void onStart(final StreamController controller) {
+ outerObserver.onStart(controller);
+ }
+
+ @Override
+ public void onResponse(ListChangeStreamPartitionsResponse response) {
+ RowRange rowRange =
+ RowRange.newBuilder()
+ .setStartKeyClosed(response.getPartition().getRowRange().getStartKeyClosed())
+ .setEndKeyOpen(response.getPartition().getRowRange().getEndKeyOpen())
+ .build();
+ outerObserver.onResponse(rowRange);
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ outerObserver.onError(t);
+ }
+
+ @Override
+ public void onComplete() {
+ outerObserver.onComplete();
+ }
+ }
+}
diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTests.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTests.java
index 34c9a29d71b7..fcbcc15e308e 100644
--- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTests.java
+++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/BigtableDataClientTests.java
@@ -24,6 +24,7 @@
import com.google.api.gax.rpc.ResponseObserver;
import com.google.api.gax.rpc.ServerStreamingCallable;
import com.google.api.gax.rpc.UnaryCallable;
+import com.google.bigtable.v2.RowRange;
import com.google.cloud.bigtable.data.v2.models.BulkMutation;
import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
import com.google.cloud.bigtable.data.v2.models.Filters.Filter;
@@ -79,6 +80,9 @@ public class BigtableDataClientTests {
@Mock private Batcher mockBulkMutationBatcher;
@Mock private Batcher mockBulkReadRowsBatcher;
+ @Mock(answer = Answers.RETURNS_DEEP_STUBS)
+ private ServerStreamingCallable mockListChangeStreamPartitionsCallable;
+
private BigtableDataClient bigtableDataClient;
@Before
@@ -153,6 +157,14 @@ public void proxyReadRowsCallableTest() {
assertThat(bigtableDataClient.readRowsCallable()).isSameInstanceAs(mockReadRowsCallable);
}
+ @Test
+ public void proxyListChangeStreamPartitionsCallableTest() {
+ Mockito.when(mockStub.listChangeStreamPartitionsCallable())
+ .thenReturn(mockListChangeStreamPartitionsCallable);
+ assertThat(bigtableDataClient.listChangeStreamPartitionsCallable())
+ .isSameInstanceAs(mockListChangeStreamPartitionsCallable);
+ }
+
@Test
public void proxyReadRowAsyncTest() {
Mockito.when(mockStub.readRowCallable()).thenReturn(mockReadRowCallable);
@@ -300,6 +312,28 @@ public void proxyReadRowsAsyncTest() {
Mockito.verify(mockReadRowsCallable).call(query, mockObserver);
}
+ @Test
+ public void proxyListChangeStreamPartitionsSyncTest() {
+ Mockito.when(mockStub.listChangeStreamPartitionsCallable())
+ .thenReturn(mockListChangeStreamPartitionsCallable);
+
+ bigtableDataClient.listChangeStreamPartitions("fake-table");
+
+ Mockito.verify(mockListChangeStreamPartitionsCallable).call("fake-table");
+ }
+
+ @Test
+ public void proxyListChangeStreamPartitionsAsyncTest() {
+ Mockito.when(mockStub.listChangeStreamPartitionsCallable())
+ .thenReturn(mockListChangeStreamPartitionsCallable);
+
+ @SuppressWarnings("unchecked")
+ ResponseObserver mockObserver = Mockito.mock(ResponseObserver.class);
+ bigtableDataClient.listChangeStreamPartitionsAsync("fake-table", mockObserver);
+
+ Mockito.verify(mockListChangeStreamPartitionsCallable).call("fake-table", mockObserver);
+ }
+
@Test
public void proxySampleRowKeysCallableTest() {
Mockito.when(mockStub.sampleRowKeysCallable()).thenReturn(mockSampleRowKeysCallable);
diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/ConvertStreamExceptionCallableTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/ConvertStreamExceptionCallableTest.java
new file mode 100644
index 000000000000..8f08e15b235b
--- /dev/null
+++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/ConvertStreamExceptionCallableTest.java
@@ -0,0 +1,76 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.stub;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import com.google.api.gax.grpc.GrpcStatusCode;
+import com.google.api.gax.rpc.ApiCallContext;
+import com.google.api.gax.rpc.ApiException;
+import com.google.api.gax.rpc.InternalException;
+import com.google.api.gax.rpc.ResponseObserver;
+import com.google.api.gax.rpc.ServerStreamingCallable;
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class ConvertStreamExceptionCallableTest {
+
+ @Test
+ public void rstStreamExceptionConvertedToRetryableTest() {
+ ApiException originalException =
+ new InternalException(
+ new StatusRuntimeException(
+ Status.INTERNAL.withDescription(
+ "INTERNAL: HTTP/2 error code: INTERNAL_ERROR\nReceived Rst Stream")),
+ GrpcStatusCode.of(Status.Code.INTERNAL),
+ false);
+ assertFalse(originalException.isRetryable());
+ SettableExceptionCallable settableExceptionCallable =
+ new SettableExceptionCallable<>(originalException);
+ ConvertStreamExceptionCallable convertStreamExceptionCallable =
+ new ConvertStreamExceptionCallable<>(settableExceptionCallable);
+
+ Throwable actualError = null;
+ try {
+ convertStreamExceptionCallable.all().call("fake-request");
+ } catch (Throwable t) {
+ actualError = t;
+ }
+ assert actualError instanceof InternalException;
+ InternalException actualException = (InternalException) actualError;
+ assertTrue(actualException.isRetryable());
+ }
+
+ private static final class SettableExceptionCallable
+ extends ServerStreamingCallable {
+ private final Throwable throwable;
+
+ public SettableExceptionCallable(Throwable throwable) {
+ this.throwable = throwable;
+ }
+
+ @Override
+ public void call(
+ RequestT request, ResponseObserver responseObserver, ApiCallContext context) {
+ responseObserver.onError(throwable);
+ }
+ }
+}
diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java
index c4e5ea2e4076..32ab93d1f2dd 100644
--- a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java
+++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/EnhancedBigtableStubSettingsTest.java
@@ -701,6 +701,7 @@ public void isRefreshingChannelFalseValueTest() {
"bulkReadRowsSettings",
"checkAndMutateRowSettings",
"readModifyWriteRowSettings",
+ "listChangeStreamPartitionsSettings",
};
@Test
diff --git a/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/changestream/ListChangeStreamPartitionsUserCallableTest.java b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/changestream/ListChangeStreamPartitionsUserCallableTest.java
new file mode 100644
index 000000000000..03db35f8d6b0
--- /dev/null
+++ b/google-cloud-bigtable/src/test/java/com/google/cloud/bigtable/data/v2/stub/changestream/ListChangeStreamPartitionsUserCallableTest.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2018 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.stub.changestream;
+
+import static com.google.common.truth.Truth.assertThat;
+
+import com.google.bigtable.v2.ListChangeStreamPartitionsRequest;
+import com.google.bigtable.v2.ListChangeStreamPartitionsResponse;
+import com.google.bigtable.v2.RowRange;
+import com.google.bigtable.v2.StreamPartition;
+import com.google.cloud.bigtable.data.v2.internal.NameUtil;
+import com.google.cloud.bigtable.data.v2.internal.RequestContext;
+import com.google.cloud.bigtable.gaxx.testing.FakeStreamingApi;
+import com.google.common.collect.Lists;
+import com.google.common.truth.Truth;
+import com.google.protobuf.ByteString;
+import java.util.List;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class ListChangeStreamPartitionsUserCallableTest {
+ private final RequestContext requestContext =
+ RequestContext.create("my-project", "my-instance", "my-profile");
+
+ @Test
+ public void requestIsCorrect() {
+ FakeStreamingApi.ServerStreamingStashCallable<
+ ListChangeStreamPartitionsRequest, ListChangeStreamPartitionsResponse>
+ inner = new FakeStreamingApi.ServerStreamingStashCallable<>(Lists.newArrayList());
+ ListChangeStreamPartitionsUserCallable listChangeStreamPartitionsUserCallable =
+ new ListChangeStreamPartitionsUserCallable(inner, requestContext);
+
+ listChangeStreamPartitionsUserCallable.all().call("my-table");
+ assertThat(inner.getActualRequest())
+ .isEqualTo(
+ ListChangeStreamPartitionsRequest.newBuilder()
+ .setTableName(
+ NameUtil.formatTableName(
+ requestContext.getProjectId(), requestContext.getInstanceId(), "my-table"))
+ .setAppProfileId(requestContext.getAppProfileId())
+ .build());
+ }
+
+ @Test
+ public void responseIsConverted() {
+ FakeStreamingApi.ServerStreamingStashCallable<
+ ListChangeStreamPartitionsRequest, ListChangeStreamPartitionsResponse>
+ inner =
+ new FakeStreamingApi.ServerStreamingStashCallable<>(
+ Lists.newArrayList(
+ ListChangeStreamPartitionsResponse.newBuilder()
+ .setPartition(
+ StreamPartition.newBuilder()
+ .setRowRange(
+ RowRange.newBuilder()
+ .setStartKeyClosed(ByteString.copyFromUtf8("apple"))
+ .setEndKeyOpen(ByteString.copyFromUtf8("banana"))
+ .build())
+ .build())
+ .build()));
+ ListChangeStreamPartitionsUserCallable listChangeStreamPartitionsUserCallable =
+ new ListChangeStreamPartitionsUserCallable(inner, requestContext);
+
+ List results = listChangeStreamPartitionsUserCallable.all().call("my-table");
+ Truth.assertThat(results)
+ .containsExactly(
+ RowRange.newBuilder()
+ .setStartKeyClosed(ByteString.copyFromUtf8("apple"))
+ .setEndKeyOpen(ByteString.copyFromUtf8("banana"))
+ .build());
+ }
+}