Merge pull request #25153 from tonytanger/initial_cdc

Initial commit of boilerplate of change stream pipeline for bigtable
apache · Jan 26, 2023 · f565a94 · f565a94
2 parents 6676646 + 67b5b36
commit f565a94
Show file tree

Hide file tree

Showing 30 changed files with 2,194 additions and 12 deletions.
diff --git a/...ogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableConfig.java b/...ogle-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigtable/BigtableConfig.java
@@ -37,16 +37,19 @@
 @SuppressWarnings({
   "nullness" // TODO(https://github.com/apache/beam/issues/20497)
 })
-abstract class BigtableConfig implements Serializable {
+public abstract class BigtableConfig implements Serializable {
 
   /** Returns the project id being written to. */
-  abstract @Nullable ValueProvider<String> getProjectId();
+  public abstract @Nullable ValueProvider<String> getProjectId();
 
   /** Returns the instance id being written to. */
-  abstract @Nullable ValueProvider<String> getInstanceId();
+  public abstract @Nullable ValueProvider<String> getInstanceId();
 
   /** Returns the table being read from. */
-  abstract @Nullable ValueProvider<String> getTableId();
+  public abstract @Nullable ValueProvider<String> getTableId();
+
+  /** Returns the app profile being read from. */
+  public abstract @Nullable ValueProvider<String> getAppProfileId();
 
   /**
    * Returns the Google Cloud Bigtable instance being written to, and other parameters.
@@ -84,6 +87,8 @@ abstract static class Builder {
 
     abstract Builder setTableId(ValueProvider<String> tableId);
 
+    abstract Builder setAppProfileId(ValueProvider<String> appProfileId);
+
     /** @deprecated will be replaced by bigtable options configurator. */
     @Deprecated
     abstract Builder setBigtableOptions(BigtableOptions options);
@@ -100,46 +105,51 @@ abstract Builder setBigtableOptionsConfigurator(
     abstract BigtableConfig build();
   }
 
-  BigtableConfig withProjectId(ValueProvider<String> projectId) {
+  public BigtableConfig withProjectId(ValueProvider<String> projectId) {
     checkArgument(projectId != null, "Project Id of BigTable can not be null");
     return toBuilder().setProjectId(projectId).build();
   }
 
-  BigtableConfig withInstanceId(ValueProvider<String> instanceId) {
+  public BigtableConfig withInstanceId(ValueProvider<String> instanceId) {
     checkArgument(instanceId != null, "Instance Id of BigTable can not be null");
     return toBuilder().setInstanceId(instanceId).build();
   }
 
-  BigtableConfig withTableId(ValueProvider<String> tableId) {
+  public BigtableConfig withTableId(ValueProvider<String> tableId) {
     checkArgument(tableId != null, "tableId can not be null");
     return toBuilder().setTableId(tableId).build();
   }
 
+  public BigtableConfig withAppProfileId(ValueProvider<String> appProfileId) {
+    checkArgument(appProfileId != null, "tableId can not be null");
+    return toBuilder().setAppProfileId(appProfileId).build();
+  }
+
   /** @deprecated will be replaced by bigtable options configurator. */
   @Deprecated
-  BigtableConfig withBigtableOptions(BigtableOptions options) {
+  public BigtableConfig withBigtableOptions(BigtableOptions options) {
     checkArgument(options != null, "Bigtable options can not be null");
     return toBuilder().setBigtableOptions(options).build();
   }
 
-  BigtableConfig withBigtableOptionsConfigurator(
+  public BigtableConfig withBigtableOptionsConfigurator(
       SerializableFunction<BigtableOptions.Builder, BigtableOptions.Builder> configurator) {
     checkArgument(configurator != null, "configurator can not be null");
     return toBuilder().setBigtableOptionsConfigurator(configurator).build();
   }
 
-  BigtableConfig withValidate(boolean isEnabled) {
+  public BigtableConfig withValidate(boolean isEnabled) {
     return toBuilder().setValidate(isEnabled).build();
   }
 
   @VisibleForTesting
-  BigtableConfig withBigtableService(BigtableService bigtableService) {
+  public BigtableConfig withBigtableService(BigtableService bigtableService) {
     checkArgument(bigtableService != null, "bigtableService can not be null");
     return toBuilder().setBigtableService(bigtableService).build();
   }
 
   @VisibleForTesting
-  BigtableConfig withEmulator(String emulatorHost) {
+  public BigtableConfig withEmulator(String emulatorHost) {
     checkArgument(emulatorHost != null, "emulatorHost can not be null");
     return toBuilder().setEmulatorHost(emulatorHost).build();
   }
@@ -173,6 +183,9 @@ void populateDisplayData(DisplayData.Builder builder) {
         .addIfNotNull(
             DisplayData.item("instanceId", getInstanceId()).withLabel("Bigtable Instance Id"))
         .addIfNotNull(DisplayData.item("tableId", getTableId()).withLabel("Bigtable Table Id"))
+        .addIfNotNull(
+            DisplayData.item("appProfileId", getAppProfileId())
+                .withLabel("Bigtable App Profile Id"))
         .add(DisplayData.item("withValidation", getValidate()).withLabel("Check is table exists"));
 
     if (getBigtableOptions() != null) {
@@ -250,6 +263,7 @@ public final String toString() {
         .add("projectId", getProjectId())
         .add("instanceId", getInstanceId())
         .add("tableId", getTableId())
+        .add("appProfileId", getAppProfileId())
         .add(
             "bigtableOptionsConfigurator",
             getBigtableOptionsConfigurator() == null