Exclude forward transaction from change streams (#2601)

Author: shreyakhajanchi

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

@@ -383,6 +383,7 @@ private void processSingleDatabaseTransaction(
         .getDatabaseClient()
         .readWriteTransaction(
             Options.tag(getTxnTag(c.getPipelineOptions())),
+            Options.excludeTxnFromChangeStreams(),
             Options.priority(spannerConfig.getRpcPriority().get()))
         .run(
             (TransactionRunner.TransactionCallable<Void>)
@@ -439,6 +440,7 @@ void processCrossDatabaseTransaction(
         .getDatabaseClient()
         .readWriteTransaction(
             Options.tag(getTxnTag(c.getPipelineOptions())),
+            Options.excludeTxnFromChangeStreams(),
             Options.priority(spannerConfig.getRpcPriority().get()))
         .allowNestedTransaction()
         .run(
@@ -463,6 +465,7 @@ void processCrossDatabaseTransaction(
                       .getDatabaseClient()
                       .readWriteTransaction(
                           Options.tag(getTxnTag(c.getPipelineOptions())),
+                          Options.excludeTxnFromChangeStreams(),
                           Options.priority(spannerConfig.getRpcPriority().get()))
                       .run(
                           (TransactionRunner.TransactionCallable<Void>)

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

@@ -166,7 +166,8 @@ public void testProcessElement() {
               TransactionRunner.TransactionCallable<Void> callable = invocation.getArgument(0);
               return callable.run(transactionContext);
             });
-    when(databaseClientMock.readWriteTransaction(any(), any())).thenReturn(transactionCallableMock);
+    when(databaseClientMock.readWriteTransaction(any(), any(), any()))
+        .thenReturn(transactionCallableMock);
 
     SpannerTransactionWriterDoFn spannerTransactionWriterDoFn =
         new SpannerTransactionWriterDoFn(
@@ -307,7 +308,8 @@ public void testProcessElementWithRetryableSpannerException() {
               throw SpannerExceptionFactory.newSpannerException(
                   ErrorCode.ABORTED, "Transaction Aborted");
             });
-    when(databaseClientMock.readWriteTransaction(any(), any())).thenReturn(transactionCallableMock);
+    when(databaseClientMock.readWriteTransaction(any(), any(), any()))
+        .thenReturn(transactionCallableMock);
 
     SpannerTransactionWriterDoFn spannerTransactionWriterDoFn =
         new SpannerTransactionWriterDoFn(
@@ -372,7 +374,8 @@ public void testProcessElementWithNonRetryableSpannerException() {
               throw SpannerExceptionFactory.newSpannerException(
                   ErrorCode.FAILED_PRECONDITION, "title must not be NULL in table Books");
             });
-    when(databaseClientMock.readWriteTransaction(any(), any())).thenReturn(transactionCallableMock);
+    when(databaseClientMock.readWriteTransaction(any(), any(), any()))
+        .thenReturn(transactionCallableMock);
 
     SpannerTransactionWriterDoFn spannerTransactionWriterDoFn =
         new SpannerTransactionWriterDoFn(

v2/spanner-to-sourcedb/README.md

@@ -109,7 +109,8 @@ A few prerequisites must be considered before starting with reverse replication.
   CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (
   retention_period = '7d',
-  value_capture_type = 'NEW_ROW'
+  value_capture_type = 'NEW_ROW',
+  allow_txn_exclusion = true
   );
   ```
 15. The Dataflow template creates a pool of database connections per Dataflow worker. The maxShardConnections template parameter, defaulting to 10,000 represents the maximum connections allowed for a given database. The maxWorkers Dataflow configuration should not exceed the maxShardConnections value, else the template launch will fail as we do not want to overload the database.

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDbITBase.java

@@ -358,7 +358,7 @@ protected SpannerResourceManager createSpannerDBAndTableWithNColumns(
     }
 
     String ddlStream =
-        "CREATE CHANGE STREAM allstream FOR ALL OPTIONS (value_capture_type = 'NEW_ROW', retention_period = '7d')";
+        "CREATE CHANGE STREAM allstream FOR ALL OPTIONS (value_capture_type = 'NEW_ROW', retention_period = '7d', allow_txn_exclusion = true)";
     try {
       spannerResourceManager.executeDdlStatement(ddlStream);
     } catch (Exception e) {

v2/spanner-to-sourcedb/src/test/resources/SpannerToCassandraSourceIT/spanner-schema.sql

@@ -237,5 +237,6 @@ CREATE TABLE testtable_03tpcovf16ed0klxm3v808ch3btgq0uk (
 CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (
   value_capture_type = 'NEW_ROW',
-  retention_period = '7d'
+  retention_period = '7d',
+  allow_txn_exclusion = true
 );

v2/spanner-to-sourcedb/src/test/resources/SpannerToCassandraSourceIT/spanner-transformation-schema.sql

@@ -10,5 +10,6 @@ CREATE TABLE IF NOT EXISTS customers (
 CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (
   value_capture_type = 'NEW_ROW',
-  retention_period = '7d'
+  retention_period = '7d',
+  allow_txn_exclusion = true
 );

v2/spanner-to-sourcedb/src/test/resources/SpannerToCassandraSourceLT/spanner-schema.sql

@@ -11,5 +11,6 @@ CREATE TABLE `person` (
 CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (
   value_capture_type = 'NEW_ROW',
-  retention_period = '7d'
+  retention_period = '7d',
+  allow_txn_exclusion = true
 );

v2/spanner-to-sourcedb/src/test/resources/SpannerToMySqlSourceLT/spanner-schema.sql

@@ -11,5 +11,6 @@ CREATE TABLE IF NOT EXISTS `Person` (
 CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (
   value_capture_type = 'NEW_ROW',
-  retention_period = '7d'
+  retention_period = '7d',
+  allow_txn_exclusion = true
 );

v2/spanner-to-sourcedb/src/test/resources/SpannerToSourceDbCustomShardIT/spanner-schema.sql

@@ -6,5 +6,6 @@ CREATE TABLE IF NOT EXISTS Singers (
 CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (
   value_capture_type = 'NEW_ROW',
-  retention_period = '7d'
+  retention_period = '7d',
+  allow_txn_exclusion = true
 );

v2/spanner-to-sourcedb/src/test/resources/SpannerToSourceDbCustomTransformationIT/spanner-schema.sql

@@ -30,5 +30,6 @@ CREATE TABLE IF NOT EXISTS AllDatatypeTransformation (
 CREATE CHANGE STREAM allstream
   FOR ALL OPTIONS (
   value_capture_type = 'NEW_ROW',
-  retention_period = '7d'
+  retention_period = '7d',
+  allow_txn_exclusion = true
 );