bug-fix: Use jdbc connection properties for reverse migration

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/utils/ShardFileReaderTest.java

@@ -45,8 +45,25 @@ public void shardFileReading() {
     List<Shard> expectedShards =
         Arrays.asList(
             new Shard(
-                "shardA", "hostShardA", "3306", "test", "test", "test", "namespaceA", null, null),
-            new Shard("shardB", "hostShardB", "3306", "test", "test", "test", null, null, null));
+                "shardA",
+                "hostShardA",
+                "3306",
+                "test",
+                "test",
+                "test",
+                "namespaceA",
+                null,
+                "jdbcCompliantTruncation=true"),
+            new Shard(
+                "shardB",
+                "hostShardB",
+                "3306",
+                "test",
+                "test",
+                "test",
+                null,
+                null,
+                "jdbcCompliantTruncation=true"));
 
     assertEquals(shards, expectedShards);
   }

v2/spanner-common/src/test/resources/shard.json

@@ -6,14 +6,16 @@
   "password":"test",
   "port":"3306",
   "dbName":"test",
-  "namespace": "namespaceA"
+  "namespace": "namespaceA",
+  "connectionProperties": "jdbcCompliantTruncation=true"
   },
  {
   "logicalShardId":"shardB",
   "host":"hostShardB",
   "user":"test",
   "password":"test",
   "port":"3306",
-  "dbName":"test"
+  "dbName":"test",
+  "connectionProperties": "jdbcCompliantTruncation=true"
    }
 ]

v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/connection/JdbcConnectionHelper.java

@@ -62,9 +62,8 @@
       config.setMaximumPoolSize(connectionHelperRequest.getMaxConnections());
       config.setConnectionInitSql(connectionHelperRequest.getConnectionInitQuery());
       Properties jdbcProperties = new Properties();
-      if (connectionHelperRequest.getProperties() != null
-          && !connectionHelperRequest.getProperties().isEmpty()) {
-        try (StringReader reader = new StringReader(connectionHelperRequest.getProperties())) {
+      if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) {
+        try (StringReader reader = new StringReader(shard.getConnectionProperties())) {
           jdbcProperties.load(reader);
         } catch (IOException e) {
           LOG.error("Error converting string to properties: {}", e.getMessage());

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/dbutils/connection/JdbcConnectionHelperTest.java

@@ -19,18 +19,26 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard;
 import com.google.cloud.teleport.v2.templates.exceptions.ConnectionException;
+import com.google.cloud.teleport.v2.templates.models.ConnectionHelperRequest;
+import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 import java.sql.Connection;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.mockito.Mock;
+import org.mockito.MockedConstruction;
 import org.mockito.MockitoAnnotations;
 
 @RunWith(JUnit4.class)
@@ -41,10 +49,13 @@ public class JdbcConnectionHelperTest {
 
   @Mock private Connection mockConnection;
 
+  @Mock private HikariConfig mockHikariConfig;
+
   @Before
   public void setUp() {
     MockitoAnnotations.openMocks(this);
     connectionHelper = new JdbcConnectionHelper();
+    mockDataSource = mock(HikariDataSource.class);
   }
 
   @Test
@@ -76,4 +87,31 @@ public void testGetConnectionPoolNotFound() throws ConnectionException {
     connectionHelper.setConnectionPoolMap(Map.of());
     assertNull(connectionHelper.getConnection("invalid-key"));
   }
+
+  @Test
+  public void testInitConnectionPool() {
+    ConnectionHelperRequest mockRequest = mock(ConnectionHelperRequest.class);
+    Shard mockShard = mock(Shard.class);
+    when(mockShard.getHost()).thenReturn("localhost");
+    when(mockShard.getPort()).thenReturn("3306");
+    when(mockShard.getDbName()).thenReturn("testdb");
+    when(mockShard.getUserName()).thenReturn("testuser");
+    when(mockShard.getPassword()).thenReturn("testpassword");
+    when(mockShard.getConnectionProperties()).thenReturn("useSSL=false");
+
+    List<Shard> mockShards = Collections.singletonList(mockShard);
+    when(mockRequest.getShards()).thenReturn(mockShards);
+    when(mockRequest.getDriver()).thenReturn("com.mysql.cj.jdbc.Driver");
+    when(mockRequest.getMaxConnections()).thenReturn(10);
+    when(mockRequest.getConnectionInitQuery()).thenReturn("SELECT 1");
+
+    try (MockedConstruction<HikariDataSource> mockedConstruction =
+        mockConstruction(
+            HikariDataSource.class,
+            (mock, context) -> when(mock.getConnection()).thenReturn(mock(Connection.class)))) {
+      connectionHelper.init(mockRequest);
+
+      assertTrue(connectionHelper.isConnectionPoolInitialized());
+    }
+  }
 }

v2/spanner-to-sourcedb/src/test/resources/shard.json

@@ -5,14 +5,16 @@
   "user":"test",
   "password":"test",
   "port":"3306",
-  "dbName":"test"
+  "dbName":"test",
+  "connectionProperties": "useSSL=true"
   },
  {
   "logicalShardId":"shardB",
   "host":"hostShardB",
   "user":"test",
   "password":"test",
   "port":"3306",
-  "dbName":"test"
+  "dbName":"test",
+  "connectionProperties": "useSSL=true"
    }
 ]

v2/spanner-to-sourcedb/terraform/samples/spanner-to-sharded-mysql/README.md

@@ -350,6 +350,31 @@ below -
 ]
 ```
 
+#### Specifying connection properties for JDBC
+
+```json
+[
+    {
+    "logicalShardId": "shard1",
+    "host": "10.11.12.13",
+    "user": "root",
+    "password":"<YOUR_PWD_HERE>",
+    "port": "3306",
+    "dbName": "db1", 
+    "connectionProperties": "jdbcCompliantTruncation=true"
+    },
+    {
+    "logicalShardId": "shard2",
+    "host": "10.11.12.14",
+    "user": "root",
+    "password":"<YOUR_PWD_HERE>",
+    "port": "3306",
+    "dbName": "db2", 
+    "connectionProperties": "jdbcCompliantTruncation=true"
+    }
+]
+```
+
 This file is generated from the configuration provided in the `var.shard_list`
 parameter.
 

v2/spanner-to-sourcedb/terraform/samples/spanner-to-sharded-mysql/variables.tf

@@ -68,12 +68,13 @@ variable "dataflow_params" {
 variable "shard_list" {
   description = "Details of the source shards to do the reverse replication to"
   type = list(object({
-    logicalShardId   = string
-    host             = string
-    user             = string
-    secretManagerUri = optional(string)
-    password         = optional(string)
-    port             = string
-    dbName           = string
+    logicalShardId       = string
+    host                 = string
+    user                 = string
+    secretManagerUri     = optional(string)
+    password             = optional(string)
+    port                 = string
+    dbName               = string
+    connectionProperties = optional(string)
   }))
 }