feat: Add readChangeStream callables

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

@@ -895,8 +895,9 @@ public Map<String, String> extract(
    *       dispatch the RPC.
    *   <li>Upon receiving the response stream, it will produce a stream of ChangeStreamRecordT. In
    *       case of mutations, it will merge the {@link ReadChangeStreamResponse.DataChange}s into
-   *       logical mutations. The actual change stream record implementation can be configured by
-   *       the {@code changeStreamRecordAdapter} parameter.
+   *       {@link com.google.cloud.bigtable.data.v2.models.ChangeStreamMutation}. The actual change
+   *       stream record implementation can be configured by the {@code changeStreamRecordAdapter}
+   *       parameter.
    *   <li>TODO: Retry/resume on failure.
    *   <li>Add tracing & metrics.
    * </ul>
@@ -905,42 +906,6 @@ public Map<String, String> extract(
       ServerStreamingCallable<ReadChangeStreamQuery, ChangeStreamRecordT>
           createReadChangeStreamCallable(
               ChangeStreamRecordAdapter<ChangeStreamRecordT> changeStreamRecordAdapter) {
-    ServerStreamingCallable<ReadChangeStreamRequest, ChangeStreamRecordT> readChangeStreamCallable =
-        createReadChangeStreamBaseCallable(
-            settings.readChangeStreamSettings(), changeStreamRecordAdapter);
-
-    ServerStreamingCallable<ReadChangeStreamQuery, ChangeStreamRecordT>
-        readChangeStreamUserCallable =
-            new ReadChangeStreamUserCallable<>(readChangeStreamCallable, requestContext);
-
-    SpanName span = getSpanName("ReadChangeStream");
-    ServerStreamingCallable<ReadChangeStreamQuery, ChangeStreamRecordT> traced =
-        new TracedServerStreamingCallable<>(
-            readChangeStreamUserCallable, clientContext.getTracerFactory(), span);
-
-    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
-  }
-
-  /**
-   * Creates a callable chain to handle ReadRows RPCs. The chain will:
-   *
-   * <ul>
-   *   <li>Dispatch the RPC with {@link ReadChangeStreamRequest}.
-   *   <li>Upon receiving the response stream, it will produce a stream of ChangeStreamRecordT. In
-   *       case of mutations, it will merge the {@link ReadChangeStreamResponse.DataChange}s into
-   *       logical mutations. The actual change stream record implementation can be configured by
-   *       the {@code changeStreamRecordAdapter} parameter.
-   *   <li>Add header tracer for tracking GFE metrics.
-   *   <li>TODO: Retry/resume on failure.
-   * </ul>
-   *
-   * <p>NOTE: the caller is responsible for adding tracing & metrics.
-   */
-  private <ReqT, ChangeStreamRecordT>
-      ServerStreamingCallable<ReadChangeStreamRequest, ChangeStreamRecordT>
-          createReadChangeStreamBaseCallable(
-              ServerStreamingCallSettings<ReqT, ChangeStreamRecord> readChangeStreamSettings,
-              ChangeStreamRecordAdapter<ChangeStreamRecordT> changeStreamRecordAdapter) {
     ServerStreamingCallable<ReadChangeStreamRequest, ReadChangeStreamResponse> base =
         GrpcRawCallableFactory.createServerStreamingCallable(
             GrpcCallSettings.<ReadChangeStreamRequest, ReadChangeStreamResponse>newBuilder()
@@ -956,7 +921,7 @@ public Map<String, String> extract(
                       }
                     })
                 .build(),
-            readChangeStreamSettings.getRetryableCodes());
+            settings.readChangeStreamSettings().getRetryableCodes());
 
     ServerStreamingCallable<ReadChangeStreamRequest, ReadChangeStreamResponse> withStatsHeaders =
         new StatsHeadersServerStreamingCallable<>(base);
@@ -975,9 +940,9 @@ public Map<String, String> extract(
     ServerStreamingCallSettings<ReadChangeStreamRequest, ChangeStreamRecordT> innerSettings =
         ServerStreamingCallSettings.<ReadChangeStreamRequest, ChangeStreamRecordT>newBuilder()
             // TODO: setResumptionStrategy.
-            .setRetryableCodes(readChangeStreamSettings.getRetryableCodes())
-            .setRetrySettings(readChangeStreamSettings.getRetrySettings())
-            .setIdleTimeout(readChangeStreamSettings.getIdleTimeout())
+            .setRetryableCodes(settings.readChangeStreamSettings().getRetryableCodes())
+            .setRetrySettings(settings.readChangeStreamSettings().getRetrySettings())
+            .setIdleTimeout(settings.readChangeStreamSettings().getIdleTimeout())
             .build();
 
     ServerStreamingCallable<ReadChangeStreamRequest, ChangeStreamRecordT> watched =
@@ -988,7 +953,19 @@ public Map<String, String> extract(
 
     // TODO: Add ReadChangeStreamRetryCompletedCallable.
 
-    return Callables.retrying(withBigtableTracer, innerSettings, clientContext);
+    ServerStreamingCallable<ReadChangeStreamRequest, ChangeStreamRecordT> readChangeStreamCallable =
+        Callables.retrying(withBigtableTracer, innerSettings, clientContext);
+
+    ServerStreamingCallable<ReadChangeStreamQuery, ChangeStreamRecordT>
+        readChangeStreamUserCallable =
+            new ReadChangeStreamUserCallable<>(readChangeStreamCallable, requestContext);
+
+    SpanName span = getSpanName("ReadChangeStream");
+    ServerStreamingCallable<ReadChangeStreamQuery, ChangeStreamRecordT> traced =
+        new TracedServerStreamingCallable<>(
+            readChangeStreamUserCallable, clientContext.getTracerFactory(), span);
+
+    return traced.withDefaultCallContext(clientContext.getDefaultCallContext());
   }
 
   /**

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/stub/changestream/ReadChangeStreamUserCallable.java

@@ -15,14 +15,19 @@
  */
 package com.google.cloud.bigtable.data.v2.stub.changestream;
 
+import com.google.api.core.InternalApi;
 import com.google.api.gax.rpc.ApiCallContext;
 import com.google.api.gax.rpc.ResponseObserver;
 import com.google.api.gax.rpc.ServerStreamingCallable;
 import com.google.bigtable.v2.ReadChangeStreamRequest;
 import com.google.cloud.bigtable.data.v2.internal.RequestContext;
 import com.google.cloud.bigtable.data.v2.models.ReadChangeStreamQuery;
 
-/** A ServerStreamingCallable that reads a change stream given the {@link ReadChangeStreamQuery}. */
+/**
+ * A ServerStreamingCallable that converts a {@link ReadChangeStreamQuery} to a {@link
+ * ReadChangeStreamRequest}.
+ */
+@InternalApi("Used in Changestream beam pipeline.")
 public class ReadChangeStreamUserCallable<ChangeStreamRecordT>
     extends ServerStreamingCallable<ReadChangeStreamQuery, ChangeStreamRecordT> {
   private final ServerStreamingCallable<ReadChangeStreamRequest, ChangeStreamRecordT> inner;