[WIP] DML Statement PR

v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT.java → v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerShardedMigrationWithMigrationMySqlShardIdColumnIT.java

@@ -52,12 +52,12 @@
 @Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class})
 @TemplateIntegrationTest(DataStreamToSpanner.class)
 @RunWith(JUnit4.class)
-public class DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT
+public class DataStreamToSpannerShardedMigrationWithMigrationMySqlShardIdColumnIT
     extends DataStreamToSpannerITBase {
 
   private static final Logger LOG =
       LoggerFactory.getLogger(
-          DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT.class);
+          DataStreamToSpannerShardedMigrationWithMigrationMySqlShardIdColumnIT.class);
 
   private static final String TABLE = "Users";
   private static final String MOVIE_TABLE = "Movie";
@@ -75,7 +75,7 @@ public class DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT
   private static final String SPANNER_DDL_RESOURCE =
       "DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT/spanner-schema.sql";
 
-  private static HashSet<DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT>
+  private static HashSet<DataStreamToSpannerShardedMigrationWithMigrationMySqlShardIdColumnIT>
       testInstances = new HashSet<>();
   private static PipelineLauncher.LaunchInfo jobInfo1;
   private static PipelineLauncher.LaunchInfo jobInfo2;
@@ -92,7 +92,7 @@ public class DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT
   public void setUp() throws IOException, InterruptedException {
     // Prevent cleaning up of dataflow job after a test method is executed.
     skipBaseCleanup = true;
-    synchronized (DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT.class) {
+    synchronized (DataStreamToSpannerShardedMigrationWithMigrationMySqlShardIdColumnIT.class) {
       testInstances.add(this);
       if (spannerResourceManager == null) {
         spannerResourceManager = setUpSpannerResourceManager();
@@ -150,7 +150,7 @@ public void setUp() throws IOException, InterruptedException {
    */
   @AfterClass
   public static void cleanUp() throws IOException {
-    for (DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT instance : testInstances) {
+    for (DataStreamToSpannerShardedMigrationWithMigrationMySqlShardIdColumnIT instance : testInstances) {
       instance.tearDownBase();
     }
     ResourceManagerUtils.cleanResources(spannerResourceManager, pubsubResourceManager);

v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT.java → v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DataStreamToSpannerShardedMigrationWithoutMigrationMySqlShardIdColumnIT.java

@@ -51,17 +51,17 @@
 @Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class})
 @TemplateIntegrationTest(DataStreamToSpanner.class)
 @RunWith(JUnit4.class)
-public class DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT
+public class DataStreamToSpannerShardedMigrationWithoutMigrationMySqlShardIdColumnIT
     extends DataStreamToSpannerITBase {
   private static final Logger LOG =
       LoggerFactory.getLogger(
-          DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT.class);
+          DataStreamToSpannerShardedMigrationWithoutMigrationMySqlShardIdColumnIT.class);
 
   private static final String TABLE = "Users";
   private static final String SPANNER_DDL_RESOURCE =
       "DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT/spanner-schema.sql";
 
-  private static HashSet<DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT>
+  private static HashSet<DataStreamToSpannerShardedMigrationWithoutMigrationMySqlShardIdColumnIT>
       testInstances = new HashSet<>();
   private static PipelineLauncher.LaunchInfo jobInfo1;
   private static PipelineLauncher.LaunchInfo jobInfo2;
@@ -78,7 +78,7 @@ public class DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT
   public void setUp() throws IOException {
     // Prevent cleaning up of dataflow job after a test method is executed.
     skipBaseCleanup = true;
-    synchronized (DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT.class) {
+    synchronized (DataStreamToSpannerShardedMigrationWithoutMigrationMySqlShardIdColumnIT.class) {
       testInstances.add(this);
       if (spannerResourceManager == null) {
         spannerResourceManager = setUpSpannerResourceManager();
@@ -131,7 +131,7 @@ public void setUp() throws IOException {
    */
   @AfterClass
   public static void cleanUp() throws IOException {
-    for (DataStreamToSpannerShardedMigrationWithoutMigrationShardIdColumnIT instance :
+    for (DataStreamToSpannerShardedMigrationWithoutMigrationMySqlShardIdColumnIT instance :
         testInstances) {
       instance.tearDownBase();
     }

v2/datastream-to-spanner/src/test/java/com/google/cloud/teleport/v2/templates/DatastreamToSpannerSingleDFShardedMigrationIT.java

@@ -87,7 +87,7 @@ public class DatastreamToSpannerSingleDFShardedMigrationIT extends DataStreamToS
   public void setUp() throws IOException, InterruptedException {
     // Prevent cleaning up of dataflow job after a test method is executed.
     skipBaseCleanup = true;
-    synchronized (DataStreamToSpannerShardedMigrationWithMigrationShardIdColumnIT.class) {
+    synchronized (DataStreamToSpannerShardedMigrationWithMigrationMySqlShardIdColumnIT.class) {
       testInstances.add(this);
       if (spannerResourceManager == null) {
         spannerResourceManager = setUpSpannerResourceManager();

v2/gcs-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/common/ProcessingContext.java

@@ -16,15 +16,16 @@
 package com.google.cloud.teleport.v2.templates.common;
 
 import com.google.cloud.teleport.v2.spanner.migrations.schema.Schema;
-import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
+import com.google.cloud.teleport.v2.spanner.migrations.shard.IShard;
+
 import java.io.Serializable;
 import java.util.Objects;
 import org.joda.time.Duration;
 
 /** Each worker task context. */
 public class ProcessingContext implements Serializable {
 
-  private Shard shard;
+  private IShard iShard;
   private Schema schema;
 
   private String sourceDbTimezoneOffset;
@@ -34,14 +35,14 @@ public class ProcessingContext implements Serializable {
   private String runId;
 
   public ProcessingContext(
-      Shard shard,
+      IShard iShard,
       Schema schema,
       String sourceDbTimezoneOffset,
       String startTimestamp,
       Duration windowDuration,
       String gcsPath,
       String runId) {
-    this.shard = shard;
+    this.iShard = iShard;
     this.schema = schema;
     this.sourceDbTimezoneOffset = sourceDbTimezoneOffset;
     this.startTimestamp = startTimestamp;
@@ -50,8 +51,8 @@ public ProcessingContext(
     this.runId = runId;
   }
 
-  public Shard getShard() {
-    return shard;
+  public IShard getShard() {
+    return iShard;
   }
 
   public Schema getSchema() {
@@ -85,8 +86,8 @@ public String getRunId() {
   @Override
   public String toString() {
 
-    return "{ Shard details :"
-        + shard.toString()
+    return "{ IShard details :"
+        + iShard.toString()
         + " sourceDbTimezoneOffset: "
         + sourceDbTimezoneOffset
         + " startTimestamp: "

v2/gcs-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/constants/Constants.java

@@ -33,12 +33,12 @@ public class Constants {
   /** Run mode - resumeAll. */
   public static final String RUN_MODE_RESUME_ALL = "resumeAll";
 
-  /** Shard progress status - success. */
+  /** IShard progress status - success. */
   public static final String SHARD_PROGRESS_STATUS_SUCCESS = "SUCCESS";
 
-  /** Shard progress status - error. */
+  /** IShard progress status - error. */
   public static final String SHARD_PROGRESS_STATUS_ERROR = "ERROR";
 
-  /** Shard progress status - reprocess. */
+  /** IShard progress status - reprocess. */
   public static final String SHARD_PROGRESS_STATUS_REPROCESS = "REPROCESS";
 }

v2/gcs-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/processing/handler/GCSToSourceStreamingHandler.java

@@ -57,7 +57,7 @@ public static String process(
       List<TrimmedShardedDataChangeRecord> records = inputFileReader.getRecords();
       Instant readEndTime = Instant.now();
       LOG.info(
-          "Shard "
+          "IShard "
               + shardId
               + ": read "
               + records.size()
@@ -82,7 +82,7 @@ public static String process(
       MySqlDao dao =
           new DaoFactory(
                   connectString,
-                  taskContext.getShard().getUserName(),
+                  taskContext.getShard().getUser(),
                   taskContext.getShard().getPassword())
               .getMySqlDao(shardId);
 
@@ -101,7 +101,7 @@ public static String process(
       markShardSuccess(taskContext, spannerDao, fileProcessedStartInterval);
       dao.cleanup();
       LOG.info(
-          "Shard " + shardId + ": Successfully processed batch of " + records.size() + " records.");
+          "IShard " + shardId + ": Successfully processed batch of " + records.size() + " records.");
     } catch (Exception e) {
       Metrics.counter(GCSToSourceStreamingHandler.class, "shard_failed_" + shardId).inc();
       markShardFailure(taskContext, spannerDao, fileProcessedStartInterval);

v2/gcs-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/processing/handler/InputRecordProcessor.java

@@ -138,7 +138,7 @@ public static void processRecords(
       dao.batchWrite(dmlBatch);
       Instant daoEndTime = Instant.now();
       LOG.info(
-          "Shard "
+          "MySqlShard "
               + shardId
               + ": Write to mysql for "
               + recordList.size()

v2/gcs-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/transforms/GcsToSourceStreamer.java

@@ -182,7 +182,7 @@ public void onExpiry(
       @StateId("stopProcessing") ValueState<Boolean> stopProcessing) {
     String shardId = keyString.read();
     LOG.info(
-        "Shard " + shardId + ": started timer processing for expiry time: " + context.timestamp());
+        "IShard " + shardId + ": started timer processing for expiry time: " + context.timestamp());
     ProcessingContext taskContext = processingContext.read();
     Boolean failedShard = stopProcessing.read();
     if (failedShard != null && failedShard) {

v2/gcs-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/utils/ProcessingContextGenerator.java

@@ -17,7 +17,7 @@
 
 import com.google.cloud.teleport.v2.spanner.migrations.metadata.SpannerToGcsJobMetadata;
 import com.google.cloud.teleport.v2.spanner.migrations.schema.Schema;
-import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
+import com.google.cloud.teleport.v2.spanner.migrations.shard.IShard;
 import com.google.cloud.teleport.v2.spanner.migrations.utils.SecretManagerAccessorImpl;
 import com.google.cloud.teleport.v2.spanner.migrations.utils.SessionFileReader;
 import com.google.cloud.teleport.v2.spanner.migrations.utils.ShardFileReader;
@@ -66,7 +66,7 @@ public static Map<String, ProcessingContext> getProcessingContextForGCS(
 
     Schema schema = SessionFileReader.read(sessionFilePath);
     ShardFileReader shardFileReader = new ShardFileReader(new SecretManagerAccessorImpl());
-    List<Shard> shards = shardFileReader.getOrderedShardDetails(sourceShardsFilePath);
+    List<IShard> iShards = shardFileReader.getOrderedShardDetails(sourceShardsFilePath);
 
     ShardProgressTracker shardProgressTracker =
         new ShardProgressTracker(
@@ -92,7 +92,7 @@ public static Map<String, ProcessingContext> getProcessingContextForGCS(
               metadataDatabase,
               tableSuffix,
               runId,
-              shards,
+                  iShards,
               schema,
               isMetadataDbPostgres);
     } else {
@@ -105,7 +105,7 @@ public static Map<String, ProcessingContext> getProcessingContextForGCS(
               metadataDatabase,
               tableSuffix,
               runId,
-              shards,
+                  iShards,
               schema,
               shardProgressTracker,
               runMode,
@@ -126,7 +126,7 @@ private static Map<String, ProcessingContext> getProcessingContextForRegularMode
       String metadataDatabase,
       String tableSuffix,
       String runId,
-      List<Shard> shards,
+      List<IShard> iShards,
       Schema schema,
       boolean isMetadataDbPostgres) {
 
@@ -157,18 +157,18 @@ private static Map<String, ProcessingContext> getProcessingContextForRegularMode
 
     Duration duration = DurationUtils.parseDuration(windowDuration);
 
-    for (Shard shard : shards) {
-      LOG.info(" The sorted shard is: {}", shard);
+    for (IShard iShard : iShards) {
+      LOG.info(" The sorted IShards is: {}", iShard);
       ProcessingContext taskContext =
           new ProcessingContext(
-              shard,
+                  iShard,
               schema,
               sourceDbTimezoneOffset,
               startTimestamp,
               duration,
               gcsInputDirectoryPath,
               runId);
-      response.put(shard.getLogicalShardId(), taskContext);
+      response.put(iShard.getLogicalShardId(), taskContext);
     }
 
     return response;
@@ -182,7 +182,7 @@ private static Map<String, ProcessingContext> getProcessingContextForReprocessOr
       String metadataDatabase,
       String tableSuffix,
       String runId,
-      List<Shard> shards,
+      List<IShard> iShards,
       Schema schema,
       ShardProgressTracker shardProgressTracker,
       String runMode,
@@ -228,9 +228,9 @@ private static Map<String, ProcessingContext> getProcessingContextForReprocessOr
 
     Duration duration = DurationUtils.parseDuration(windowDuration);
 
-    for (Shard shard : shards) {
-      LOG.info(" The sorted shard is: {}", shard);
-      ShardProgress shardProgress = shardProgressList.get(shard.getLogicalShardId());
+    for (IShard iShard : iShards) {
+      LOG.info(" The sorted shards is: {}", iShard);
+      ShardProgress shardProgress = shardProgressList.get(iShard.getLogicalShardId());
 
       String shardStartTime = null;
       if (shardProgress != null) {
@@ -239,27 +239,27 @@ private static Map<String, ProcessingContext> getProcessingContextForReprocessOr
         if ((runMode.equals(Constants.RUN_MODE_RESUME_SUCCESS)
                 || runMode.equals(Constants.RUN_MODE_RESUME_ALL))
             && shardProgress.getStatus().equals(Constants.SHARD_PROGRESS_STATUS_SUCCESS)) {
-          // Advance the start time by window duration for successful shards
+          // Advance the start time by window duration for successful Shards
           shardStartTimeInst = shardStartTimeInst.plus(duration);
         }
         shardStartTime = shardStartTimeInst.toString();
 
-        LOG.info(" The startTime for shard {} is : {}", shard, shardStartTime);
+        LOG.info(" The startTime for shards {} is : {}", iShard, shardStartTime);
       } else {
         LOG.info(
-            " Skipping shard: {} as it does not qualify for given runMode {}.", shard, runMode);
+            " Skipping shards: {} as it does not qualify for given runMode {}.", iShard, runMode);
         continue;
       }
       ProcessingContext taskContext =
           new ProcessingContext(
-              shard,
+                  iShard,
               schema,
               sourceDbTimezoneOffset,
               shardStartTime,
               duration,
               gcsInputDirectoryPath,
               runId);
-      response.put(shard.getLogicalShardId(), taskContext);
+      response.put(iShard.getLogicalShardId(), taskContext);
     }
     return response;
   }

v2/gcs-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/GCSToSourceDbInterleaveIT.java

@@ -24,7 +24,7 @@
 import com.google.cloud.spanner.Mutation;
 import com.google.cloud.teleport.metadata.SkipDirectRunnerTest;
 import com.google.cloud.teleport.metadata.TemplateIntegrationTest;
-import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
+import com.google.cloud.teleport.v2.spanner.migrations.shard.MySqlShard;
 import com.google.common.io.Resources;
 import com.google.gson.Gson;
 import com.google.gson.JsonArray;
@@ -410,19 +410,19 @@ private void launchReaderDataflowJob() throws IOException {
   private void createAndUploadShardConfigToGcs(
       GcsResourceManager gcsResourceManager, MySQLResourceManager jdbcResourceManager)
       throws IOException {
-    Shard shard = new Shard();
-    shard.setLogicalShardId("Shard1");
-    shard.setUser(jdbcResourceManager.getUsername());
-    shard.setHost(jdbcResourceManager.getHost());
-    shard.setPassword(jdbcResourceManager.getPassword());
-    shard.setPort(String.valueOf(jdbcResourceManager.getPort()));
-    shard.setDbName(jdbcResourceManager.getDatabaseName());
-    JsonObject jsObj = (JsonObject) new Gson().toJsonTree(shard).getAsJsonObject();
+    MySqlShard mySqlShard = new MySqlShard();
+    mySqlShard.setLogicalShardId("Shard1");
+    mySqlShard.setUser(jdbcResourceManager.getUsername());
+    mySqlShard.setHost(jdbcResourceManager.getHost());
+    mySqlShard.setPassword(jdbcResourceManager.getPassword());
+    mySqlShard.setPort(String.valueOf(jdbcResourceManager.getPort()));
+    mySqlShard.setDbName(jdbcResourceManager.getDatabaseName());
+    JsonObject jsObj = (JsonObject) new Gson().toJsonTree(mySqlShard).getAsJsonObject();
     jsObj.remove("secretManagerUri"); // remove field secretManagerUri
     JsonArray ja = new JsonArray();
     ja.add(jsObj);
     String shardFileContents = ja.toString();
-    LOG.info("Shard file contents: {}", shardFileContents);
-    gcsResourceManager.createArtifact("input/shard.json", shardFileContents);
+    LOG.info("MySqlShard file contents: {}", shardFileContents);
+    gcsResourceManager.createArtifact("input/mySqlShard.json", shardFileContents);
   }
 }