Refactor: Decouple Datastream template from source engines (PR 2)

Introduce a pluggable ISourceConnector interface to encapsulate all source-specific
logic for MySQL, PostgreSQL, and Oracle databases within the datastream-to-spanner
template.

This replaces inline source-specific logic with a clean, decoupled design:
- Created ISourceConnector interface.
- Created MySqlSourceConnector, PostgresqlSourceConnector, and OracleSourceConnector implementing the interface.
- Refactored SourceConnectorRegistry to manually register and manage these connectors (no AutoService/SPI dependency).
- Updated SpannerTransactionWriterDoFn and ShadowTableCreator to interact with sources purely via the ISourceConnector interface.
- Deleted obsolete ChangeEventContextFactory and ChangeEventSequenceFactory classes, delegating context and sequence creation to the connectors.
- Kept all source-specific constants in DatastreamConstants (like in main) to preserve compatibility.
- Removed auto-service dependency from pom.xml.
- All changes are strictly restricted to the datastream-to-spanner template.

TAG=agy
CONV=b7fec87d-3652-43db-a0d1-64773a054415

Author: Aditya Bharadwaj

v2/datastream-to-spanner/pom.xml

@@ -109,6 +109,7 @@
             <artifactId>google-cloud-compute</artifactId>
             <scope>test</scope>
         </dependency>
+
     </dependencies>
 
     <profiles>

v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/DataStreamToSpanner.java

@@ -45,7 +45,7 @@
 import com.google.cloud.teleport.v2.spanner.migrations.utils.TransformationContextReader;
 import com.google.cloud.teleport.v2.templates.DataStreamToSpanner.Options;
 import com.google.cloud.teleport.v2.templates.constants.DatastreamToSpannerConstants;
-import com.google.cloud.teleport.v2.templates.datastream.DatastreamConstants;
+import com.google.cloud.teleport.v2.templates.datastream.source.SourceConnectorRegistry;
 import com.google.cloud.teleport.v2.templates.spanner.ProcessInformationSchema;
 import com.google.cloud.teleport.v2.templates.transform.ChangeEventTransformerDoFn;
 import com.google.cloud.teleport.v2.transforms.DLQWriteTransform;
@@ -589,12 +589,12 @@ static void validateSourceType(Options options) {
       return;
     }
     String sourceType = getSourceType(options);
-    if (!DatastreamConstants.SUPPORTED_DATASTREAM_SOURCES.contains(sourceType)) {
+    if (!SourceConnectorRegistry.getSupportedSourceTypes().contains(sourceType)) {
       throw new IllegalArgumentException(
           "Unsupported source type found: "
               + sourceType
               + ". Specify one of the following source types: "
-              + DatastreamConstants.SUPPORTED_DATASTREAM_SOURCES);
+              + SourceConnectorRegistry.getSupportedSourceTypes());
     }
     options.setDatastreamSourceType(sourceType);
   }
@@ -620,15 +620,12 @@ static String getSourceType(Options options) {
   }
 
   static String getSourceTypeFromConfig(SourceConfig sourceConfig) {
-    if (sourceConfig.getMysqlSourceConfig() != null) {
-      return DatastreamConstants.MYSQL_SOURCE_TYPE;
-    } else if (sourceConfig.getOracleSourceConfig() != null) {
-      return DatastreamConstants.ORACLE_SOURCE_TYPE;
-    } else if (sourceConfig.getPostgresqlSourceConfig() != null) {
-      return DatastreamConstants.POSTGRES_SOURCE_TYPE;
+    try {
+      return SourceConnectorRegistry.getSourceTypeFromConfig(sourceConfig);
+    } catch (IllegalArgumentException e) {
+      LOG.error("Source Connection Profile Type Not Supported", e);
+      throw e;
     }
-    LOG.error("Source Connection Profile Type Not Supported");
-    throw new IllegalArgumentException("Unsupported source connection profile type in Datastream");
   }
 
   /**

v2/datastream-to-spanner/src/main/java/com/google/cloud/teleport/v2/templates/SpannerTransactionWriterDoFn.java

@@ -38,9 +38,10 @@
 import com.google.cloud.teleport.v2.spanner.migrations.exceptions.SpannerMigrationException;
 import com.google.cloud.teleport.v2.templates.constants.DatastreamToSpannerConstants;
 import com.google.cloud.teleport.v2.templates.datastream.ChangeEventContext;
-import com.google.cloud.teleport.v2.templates.datastream.ChangeEventContextFactory;
 import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequence;
-import com.google.cloud.teleport.v2.templates.datastream.ChangeEventSequenceFactory;
+import com.google.cloud.teleport.v2.templates.datastream.DatastreamConstants;
+import com.google.cloud.teleport.v2.templates.datastream.source.ISourceConnector;
+import com.google.cloud.teleport.v2.templates.datastream.source.SourceConnectorRegistry;
 import com.google.cloud.teleport.v2.templates.spanner.DatastreamToSpannerExceptionClassifier;
 import com.google.cloud.teleport.v2.templates.spanner.DatastreamToSpannerExceptionClassifier.ErrorTag;
 import com.google.cloud.teleport.v2.templates.utils.WatchdogRunnable;
@@ -103,6 +104,8 @@ class SpannerTransactionWriterDoFn
 
   /* SpannerAccessor must be transient so that its value is not serialized at runtime. */
   private transient SpannerAccessor spannerAccessor;
+
+  private transient ISourceConnector sourceConnector;
   /* SpannerAccessor for shadow table database must be transient so that its value is not serialized at runtime. */
   private transient SpannerAccessor shadowTableSpannerAccessor;
 
@@ -216,6 +219,7 @@ public void setup() {
             : spannerAccessor;
     mapper = new ObjectMapper();
     mapper.enable(DeserializationFeature.USE_BIG_DECIMAL_FOR_FLOATS);
+    sourceConnector = SourceConnectorRegistry.getSourceConnector(sourceType);
     // Setup and start the watchdog thread.
     transactionAttemptCount = new AtomicLong(0);
     isInTransaction = new AtomicBoolean(false);
@@ -256,6 +260,19 @@ public void processElement(ProcessContext c) {
     try {
 
       JsonNode changeEvent = mapper.readTree(msg.getPayload());
+
+      // Validate source type
+      JsonNode eventSourceTypeNode = changeEvent.get(DatastreamConstants.EVENT_SOURCE_TYPE_KEY);
+      String eventSourceType = eventSourceTypeNode != null ? eventSourceTypeNode.asText() : "";
+      if (!eventSourceType.equalsIgnoreCase(sourceType)) {
+        throw new InvalidChangeEventException(
+            "Change event with invalid source. Actual("
+                + eventSourceType
+                + "), Expected("
+                + sourceType
+                + ")");
+      }
+
       migrationShardId =
           Optional.ofNullable(changeEvent.get(SHARD_ID_COLUMN_NAME))
               .map(shardIdNode -> changeEvent.get(shardIdNode.asText()).asText())
@@ -267,13 +284,12 @@ public void processElement(ProcessContext c) {
         isRetryRecord = true;
       }
       ChangeEventContext changeEventContext =
-          ChangeEventContextFactory.createChangeEventContext(
-              changeEvent, ddl, shadowTableDdl, shadowTablePrefix, sourceType);
+          getSourceConnector()
+              .createChangeEventContext(changeEvent, ddl, shadowTableDdl, shadowTablePrefix);
 
       // Sequence information for the current change event.
       ChangeEventSequence currentChangeEventSequence =
-          ChangeEventSequenceFactory.createChangeEventSequenceFromChangeEventContext(
-              changeEventContext);
+          getSourceConnector().createChangeEventSequenceFromChangeEventContext(changeEventContext);
 
       if (usesSeparateShadowTableDb) {
         processCrossDatabaseTransaction(
@@ -395,8 +411,9 @@ private void processSingleDatabaseTransaction(
                   transactionAttemptCount.incrementAndGet();
                   // Sequence information for the last change event.
                   ChangeEventSequence previousChangeEventSequence =
-                      ChangeEventSequenceFactory.createChangeEventSequenceFromShadowTable(
-                          transaction, changeEventContext, shadowDdl, false);
+                      getSourceConnector()
+                          .createChangeEventSequenceFromShadowTable(
+                              transaction, changeEventContext, shadowDdl, false);
                   /* There was a previous event recorded with a greater sequence information
                    * than current. Hence, skip the current event.
                    */
@@ -463,8 +480,12 @@ void processCrossDatabaseTransaction(
 
                   // Build lock query based on source type
                   ChangeEventSequence previousChangeEventSequence =
-                      ChangeEventSequenceFactory.createChangeEventSequenceFromShadowTable(
-                          shadowTxn, changeEventContext, shadowDdl, /* useSqlStatments= */ true);
+                      getSourceConnector()
+                          .createChangeEventSequenceFromShadowTable(
+                              shadowTxn,
+                              changeEventContext,
+                              shadowDdl,
+                              /* useSqlStatments= */ true);
 
                   if (previousChangeEventSequence != null
                       && previousChangeEventSequence.compareTo(currentChangeEventSequence) >= 0) {
@@ -493,7 +514,7 @@ void processCrossDatabaseTransaction(
                                 // this
                                 // thread gets killed.
                                 ChangeEventSequence validationSequence =
-                                    ChangeEventSequenceFactory
+                                    getSourceConnector()
                                         .createChangeEventSequenceFromShadowTable(
                                             shadowTxn,
                                             changeEventContext,
@@ -577,6 +598,17 @@ String getTxnTag(PipelineOptions options) {
     return txnTag;
   }
 
+  private ISourceConnector getSourceConnector() {
+    if (sourceConnector == null) {
+      sourceConnector = SourceConnectorRegistry.getSourceConnector(sourceType);
+    }
+    return sourceConnector;
+  }
+
+  void setSourceConnector(ISourceConnector sourceConnector) {
+    this.sourceConnector = sourceConnector;
+  }
+
   public void setMapper(ObjectMapper mapper) {
     this.mapper = mapper;
   }