Introduce and use constants for Spanner Change Streams CDC time increment and heartbeat milliseconds.

Author: stankiewicz

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java

@@ -19,9 +19,13 @@
 
 import static java.util.stream.Collectors.toList;
 import static org.apache.beam.sdk.io.gcp.spanner.MutationUtils.isPointDelete;
+import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_CDC_TIME_INCREMENT;
 import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_CHANGE_STREAM_NAME;
+import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_HEARTBEAT_MILLIS;
 import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_INCLUSIVE_END_AT;
 import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_INCLUSIVE_START_AT;
+import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_LOW_LATENCY_CDC_TIME_INCREMENT;
+import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_LOW_LATENCY_HEARTBEAT_MILLIS;
 import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_RPC_PRIORITY;
 import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.DEFAULT_WATERMARK_REFRESH_RATE;
 import static org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamsConstants.MAX_INCLUSIVE_END_AT;
@@ -537,8 +541,8 @@ public static ReadChangeStream readChangeStream() {
         .setRpcPriority(DEFAULT_RPC_PRIORITY)
         .setInclusiveStartAt(DEFAULT_INCLUSIVE_START_AT)
         .setInclusiveEndAt(DEFAULT_INCLUSIVE_END_AT)
-        .setCdcTimeIncrement(Duration.standardMinutes(2))
-        .setHeartbeatMillis(2000)
+        .setCdcTimeIncrement(DEFAULT_CDC_TIME_INCREMENT)
+        .setHeartbeatMillis(DEFAULT_HEARTBEAT_MILLIS)
         .build();
   }
 
@@ -1931,8 +1935,8 @@ public ReadChangeStream withUsingPlainTextChannel(boolean plainText) {
 
     public ReadChangeStream withLowLatency() {
       return toBuilder()
-          .setCdcTimeIncrement(Duration.standardSeconds(1))
-          .setHeartbeatMillis(100)
+          .setCdcTimeIncrement(DEFAULT_LOW_LATENCY_CDC_TIME_INCREMENT)
+          .setHeartbeatMillis(DEFAULT_LOW_LATENCY_HEARTBEAT_MILLIS)
           .build();
     }
 

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/ChangeStreamsConstants.java

@@ -49,6 +49,14 @@ public class ChangeStreamsConstants {
    */
   public static final Timestamp DEFAULT_INCLUSIVE_END_AT = MAX_INCLUSIVE_END_AT;
 
+  public static final Duration DEFAULT_CDC_TIME_INCREMENT = Duration.standardMinutes(2);
+
+  public static final int DEFAULT_HEARTBEAT_MILLIS = 2000;
+
+  public static final Duration DEFAULT_LOW_LATENCY_CDC_TIME_INCREMENT = Duration.standardSeconds(1);
+
+  public static final int DEFAULT_LOW_LATENCY_HEARTBEAT_MILLIS = 100;
+
   /** The default priority for a change stream query is {@link RpcPriority#HIGH}. */
   public static final RpcPriority DEFAULT_RPC_PRIORITY = RpcPriority.HIGH;