feat: Support Cloud Bigtable Changestream

google-cloud-bigtable-stats/pom.xml

@@ -16,6 +16,10 @@
     <version>2.17.2-SNAPSHOT</version><!-- {x-version-update:google-cloud-bigtable:current} -->
     <description>Experimental project to shade OpenCensus dependencies.</description>
 
+    <properties>
+        <cloud.monitoring.version>3.4.1</cloud.monitoring.version>
+    </properties>
+
     <dependencyManagement>
         <dependencies>
             <dependency>

google-cloud-bigtable/clirr-ignored-differences.xml

@@ -39,6 +39,11 @@
         <differenceType>8001</differenceType>
         <className>com/google/cloud/bigtable/data/v2/stub/metrics/CompositeTracerFactory</className>
     </difference>
+    <!-- InternalApi was renamed -->
+    <difference>
+        <differenceType>8001</differenceType>
+        <className>com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsConvertExceptionCallable</className>
+    </difference>
     <!-- InternalApi that was removed -->
     <difference>
         <differenceType>8001</differenceType>

google-cloud-bigtable/pom.xml

@@ -36,6 +36,10 @@
     <!-- This is used by bigtable-prod-batch-it profile to ensure that tests work on the batch endpoint.
      Also, this property will be augmented by `internal-bigtable-prod-batch-it-prop-helper` profile -->
     <bigtable.cfe-data-batch-endpoint>batch-bigtable.googleapis.com:443</bigtable.cfe-data-batch-endpoint>
+
+    <grpc.version>1.44.0</grpc.version>
+    <protobuf.version>3.19.3</protobuf.version>
+    <protoc.version>${protobuf.version}</protoc.version>
   </properties>
 
   <dependencyManagement>
@@ -593,7 +597,36 @@
   </profiles>
 
   <build>
+    <extensions>
+      <extension>
+        <groupId>kr.motd.maven</groupId>
+        <artifactId>os-maven-plugin</artifactId>
+        <version>1.6.0</version>
+      </extension>
+    </extensions>
     <plugins>
+      <plugin>
+        <groupId>org.xolstice.maven.plugins</groupId>
+        <artifactId>protobuf-maven-plugin</artifactId>
+        <version>0.6.1</version>
+        <executions>
+          <execution>
+            <goals>
+              <goal>compile</goal>
+              <goal>compile-custom</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <protocArtifact>
+            com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}
+          </protocArtifact>
+          <pluginId>grpc-java</pluginId>
+          <pluginArtifact>
+            io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}
+          </pluginArtifact>
+        </configuration>
+      </plugin>
       <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>build-helper-maven-plugin</artifactId>

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

@@ -30,11 +30,14 @@
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.cloud.bigtable.data.v2.models.BulkMutation;
+import com.google.cloud.bigtable.data.v2.models.ChangeStreamRecord;
 import com.google.cloud.bigtable.data.v2.models.ConditionalRowMutation;
 import com.google.cloud.bigtable.data.v2.models.Filters;
 import com.google.cloud.bigtable.data.v2.models.Filters.Filter;
 import com.google.cloud.bigtable.data.v2.models.KeyOffset;
 import com.google.cloud.bigtable.data.v2.models.Query;
+import com.google.cloud.bigtable.data.v2.models.Range.ByteStringRange;
+import com.google.cloud.bigtable.data.v2.models.ReadChangeStreamQuery;
 import com.google.cloud.bigtable.data.v2.models.ReadModifyWriteRow;
 import com.google.cloud.bigtable.data.v2.models.Row;
 import com.google.cloud.bigtable.data.v2.models.RowAdapter;
@@ -1489,6 +1492,298 @@ public UnaryCallable<ReadModifyWriteRow, Row> readModifyWriteRowCallable() {
     return stub.readModifyWriteRowCallable();
   }
 
+  /**
+   * Convenience method for synchronously streaming the partitions of a table. The returned
+   * ServerStream instance is not threadsafe, it can only be used from single thread.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+   *   String tableId = "[TABLE]";
+   *
+   *   try {
+   *     ServerStream<ByteStringRange> stream = bigtableDataClient.generateInitialChangeStreamPartitions(tableId);
+   *     int count = 0;
+   *
+   *     // Iterator style
+   *     for (ByteStringRange 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();
+   *   }
+   * }
+   * }</pre>
+   *
+   * @see ServerStreamingCallable For call styles.
+   */
+  @InternalApi("Used in Changestream beam pipeline.")
+  public ServerStream<ByteStringRange> generateInitialChangeStreamPartitions(String tableId) {
+    return generateInitialChangeStreamPartitionsCallable().call(tableId);
+  }
+
+  /**
+   * Convenience method for asynchronously streaming the partitions of a table.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+   *   String tableId = "[TABLE]";
+   *
+   *   bigtableDataClient.generateInitialChangeStreamPartitionsAsync(tableId, new ResponseObserver<RowRange>() {
+   *     StreamController controller;
+   *     int count = 0;
+   *
+   *     public void onStart(StreamController controller) {
+   *       this.controller = controller;
+   *     }
+   *     public void onResponse(ByteStringRange 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
+   *     }
+   *   });
+   * }
+   * }</pre>
+   */
+  @InternalApi("Used in Changestream beam pipeline.")
+  public void generateInitialChangeStreamPartitionsAsync(
+      String tableId, ResponseObserver<ByteStringRange> observer) {
+    generateInitialChangeStreamPartitionsCallable().call(tableId, observer);
+  }
+
+  /**
+   * Streams back the results of the query. The returned callable object allows for customization of
+   * api invocation.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+   *   String tableId = "[TABLE]";
+   *
+   *   // Iterator style
+   *   try {
+   *     for(ByteStringRange partition : bigtableDataClient.generateInitialChangeStreamPartitionsCallable().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<ByteStringRange> partitions = bigtableDataClient.generateInitialChangeStreamPartitionsCallable().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<ByteStringRange> partitionFuture =
+   *     bigtableDataClient.generateInitialChangeStreamPartitionsCallable().first().futureCall(tableId);
+   *
+   *   ApiFutures.addCallback(partitionFuture, new ApiFutureCallback<ByteStringRange>() {
+   *     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
+   * }
+   * }</pre>
+   *
+   * @see ServerStreamingCallable For call styles.
+   */
+  @InternalApi("Used in Changestream beam pipeline.")
+  public ServerStreamingCallable<String, ByteStringRange>
+      generateInitialChangeStreamPartitionsCallable() {
+    return stub.generateInitialChangeStreamPartitionsCallable();
+  }
+
+  /**
+   * Convenience method for synchronously streaming the results of a {@link ReadChangeStreamQuery}.
+   * The returned ServerStream instance is not threadsafe, it can only be used from single thread.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+   *   String tableId = "[TABLE]";
+   *
+   *   ReadChangeStreamQuery query = ReadChangeStreamQuery.create(tableId)
+   *          .streamPartition("START_KEY", "END_KEY")
+   *          .startTime(Timestamp.newBuilder().setSeconds(100).build());
+   *
+   *   try {
+   *     ServerStream<ChangeStreamRecord> stream = bigtableDataClient.readChangeStream(query);
+   *     int count = 0;
+   *
+   *     // Iterator style
+   *     for (ChangeStreamRecord record : stream) {
+   *       if (++count > 10) {
+   *         stream.cancel();
+   *         break;
+   *       }
+   *       // Do something with the change stream record.
+   *     }
+   *   } catch (NotFoundException e) {
+   *     System.out.println("Tried to read a non-existent table");
+   *   } catch (RuntimeException e) {
+   *     e.printStackTrace();
+   *   }
+   * }
+   * }</pre>
+   *
+   * @see ServerStreamingCallable For call styles.
+   * @see ReadChangeStreamQuery For query options.
+   */
+  @InternalApi("Used in Changestream beam pipeline.")
+  public ServerStream<ChangeStreamRecord> readChangeStream(ReadChangeStreamQuery query) {
+    return readChangeStreamCallable().call(query);
+  }
+
+  /**
+   * Convenience method for asynchronously streaming the results of a {@link ReadChangeStreamQuery}.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+   *   String tableId = "[TABLE]";
+   *
+   *   ReadChangeStreamQuery query = ReadChangeStreamQuery.create(tableId)
+   *          .streamPartition("START_KEY", "END_KEY")
+   *          .startTime(Timestamp.newBuilder().setSeconds(100).build());
+   *
+   *   bigtableDataClient.readChangeStreamAsync(query, new ResponseObserver<ChangeStreamRecord>() {
+   *     StreamController controller;
+   *     int count = 0;
+   *
+   *     public void onStart(StreamController controller) {
+   *       this.controller = controller;
+   *     }
+   *     public void onResponse(ChangeStreamRecord record) {
+   *       if (++count > 10) {
+   *         controller.cancel();
+   *         return;
+   *       }
+   *       // Do something with the change stream record.
+   *     }
+   *     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
+   *     }
+   *   });
+   * }
+   * }</pre>
+   */
+  @InternalApi("Used in Changestream beam pipeline.")
+  public void readChangeStreamAsync(
+      ReadChangeStreamQuery query, ResponseObserver<ChangeStreamRecord> observer) {
+    readChangeStreamCallable().call(query, observer);
+  }
+
+  /**
+   * Streams back the results of the query. The returned callable object allows for customization of
+   * api invocation.
+   *
+   * <p>Sample code:
+   *
+   * <pre>{@code
+   * try (BigtableDataClient bigtableDataClient = BigtableDataClient.create("[PROJECT]", "[INSTANCE]")) {
+   *   String tableId = "[TABLE]";
+   *
+   *   ReadChangeStreamQuery query = ReadChangeStreamQuery.create(tableId)
+   *          .streamPartition("START_KEY", "END_KEY")
+   *          .startTime(Timestamp.newBuilder().setSeconds(100).build());
+   *
+   *   // Iterator style
+   *   try {
+   *     for(ChangeStreamRecord record : bigtableDataClient.readChangeStreamCallable().call(query)) {
+   *       // Do something with record
+   *     }
+   *   } catch (NotFoundException e) {
+   *     System.out.println("Tried to read a non-existent table");
+   *   } catch (RuntimeException e) {
+   *     e.printStackTrace();
+   *   }
+   *
+   *   // Sync style
+   *   try {
+   *     List<ChangeStreamRecord> records = bigtableDataClient.readChangeStreamCallable().all().call(query);
+   *   } catch (NotFoundException e) {
+   *     System.out.println("Tried to read a non-existent table");
+   *   } catch (RuntimeException e) {
+   *     e.printStackTrace();
+   *   }
+   *
+   *   // Point look up
+   *   ApiFuture<ChangeStreamRecord> recordFuture =
+   *     bigtableDataClient.readChangeStreamCallable().first().futureCall(query);
+   *
+   *   ApiFutures.addCallback(recordFuture, new ApiFutureCallback<ChangeStreamRecord>() {
+   *     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(ChangeStreamRecord result) {
+   *       System.out.println("Got record: " + result);
+   *     }
+   *   }, MoreExecutors.directExecutor());
+   *
+   *   // etc
+   * }
+   * }</pre>
+   *
+   * @see ServerStreamingCallable For call styles.
+   * @see ReadChangeStreamQuery For query options.
+   */
+  @InternalApi("Used in Changestream beam pipeline.")
+  public ServerStreamingCallable<ReadChangeStreamQuery, ChangeStreamRecord>
+      readChangeStreamCallable() {
+    return stub.readChangeStreamCallable();
+  }
+
   /** Close the clients and releases all associated resources. */
   @Override
   public void close() {