[SpannerToSourceDb] Add support for URL-encoded connection properties and enable MySQL SSL based authentication

v2/sourcedb-to-spanner/README.md

@@ -65,12 +65,18 @@ mvn test
 ### Executing Template
 
 #### Required Parameters
-* **sourceConfigURL** (Source connection config file URL): The URL of the source connection config file. The file format is dependent on the source type. For Astra, it will point to an Astra connection config file ([sample](src/test/resources/SourceConfig/astra-connection-config.json)). For JDBC, it will point to a JDBC sharding config file ([sample](src/test/resources/SourceConfig/jdbc-shard-config.json)). For Cassandra, it will point to a Cassandra driver config file ([sample](src/test/resources/SourceConfig/cassandra-driver-config.conf)). This parameter is required. Refer to src/main/scripts/create_simple_shard_config.bash for steps to generate a shard configuration.
+* **sourceConfigURL** (Source connection config file URL): The URL of the source connection config file. The file format is dependent on the source type. For Astra, it will point to an Astra connection config file ([sample](src/test/resources/SourceConfig/astra-connection-config.json)). For JDBC, it will point to a JDBC sharding config file ([sample](src/test/resources/SourceConfig/jdbc-shard-config.json)). You can optionally specify a `"connectionProperties"` string in the JDBC config to configure the JDBC connection (e.g. `useSSL=true&requireSSL=true`), where keys and values can be URL-encoded if they contain special characters. For Cassandra, it will point to a Cassandra driver config file ([sample](src/test/resources/SourceConfig/cassandra-driver-config.conf)). This parameter is required. Refer to src/main/scripts/create_simple_shard_config.bash for steps to generate a shard configuration.
 * **instanceId** (Cloud Spanner Instance Id.): The destination Cloud Spanner instance.
 * **databaseId** (Cloud Spanner Database Id.): The destination Cloud Spanner database.
 * **projectId** (Cloud Spanner Project Id.): This is the name of the Cloud Spanner project.
 * **outputDirectory** (GCS path of the output directory): The GCS path of the directory where all errors and skipped events are dumped to be used during migrations
 
+**Referencing SSL Certificates**:
+To connect to a JDBC source using a custom SSL certificate (e.g. a truststore), you must first make the certificate file available to the Dataflow workers:
+1. Upload your certificate file (e.g., `truststore.jks`) to a Google Cloud Storage bucket.
+2. When launching the Dataflow template, provide the GCS path to the `--extraFilesToStage` parameter (e.g. `--extraFilesToStage="gs://<BUCKET_NAME>/truststore.jks"`). The file will be downloaded to the `/extra_files` directory on each worker.
+3. In your shards JSON configuration, set the `"connectionProperties"` to reference this local file path. For example, for MySQL you would specify the `trustCertificateKeyStoreUrl` and password: `"connectionProperties": "useSSL=true&requireSSL=true&trustCertificateKeyStoreUrl=file:/extra_files/truststore.jks&trustCertificateKeyStorePassword=my_password"`
+
 #### Optional Parameters
 * **jdbcDriverJars** (Comma-separated Cloud Storage path(s) of the JDBC driver(s)): The comma-separated list of driver JAR files. (Example: gs://your-bucket/driver_jar1.jar,gs://your-bucket/driver_jar2.jar).
 * **jdbcDriverClassName** (JDBC driver class name): The JDBC driver class name. (Example: com.mysql.jdbc.Driver).

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java

@@ -21,6 +21,7 @@
 import com.zaxxer.hikari.HikariDataSource;
 import java.io.IOException;
 import java.io.StringReader;
+import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
 import java.util.HashMap;
 import java.util.Map;
@@ -71,10 +72,13 @@ public synchronized void init(ConnectionHelperRequest connectionHelperRequest) {
       config.setMinimumIdle(0); // avoid pre-filling connections
       Properties jdbcProperties = new Properties();
       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());
+        LOG.info(
+            "Connection properties for shard {}: {}",
+            shard.getLogicalShardId(),
+            shard.getConnectionProperties());
+        Properties parsedProps = parseProperties(shard.getConnectionProperties());
+        for (String key : parsedProps.stringPropertyNames()) {
+          jdbcProperties.setProperty(key, parsedProps.getProperty(key));
         }
       }
 
@@ -111,4 +115,42 @@ public Connection getConnection(String connectionRequestKey) throws ConnectionEx
   public void setConnectionPoolMap(Map<String, HikariDataSource> inputMap) {
     connectionPoolMap = inputMap;
   }
+
+  /**
+   * Parses connection properties from a string into a {@link Properties} object.
+   *
+   * <p>Supports both newline-delimited Java properties format and URL-encoded query parameters
+   * (separated by '&' or ';'). URL-encoded values are automatically decoded.
+   *
+   * @param connectionProperties The connection properties string.
+   * @return A Properties object containing the parsed key-value pairs.
+   */
+  public static Properties parseProperties(String connectionProperties) {
+    Properties jdbcProperties = new Properties();
+    if (connectionProperties == null || connectionProperties.isEmpty()) {
+      return jdbcProperties;
+    }
+
+    if (connectionProperties.contains("&") || connectionProperties.contains(";")) {
+      String[] pairs = connectionProperties.split("[&;]");
+      for (String pair : pairs) {
+        String[] kv = pair.split("=", 2);
+        if (kv.length == 2) {
+          String decodedKey = java.net.URLDecoder.decode(kv[0], StandardCharsets.UTF_8);
+          String decodedValue = java.net.URLDecoder.decode(kv[1], StandardCharsets.UTF_8);
+          jdbcProperties.setProperty(decodedKey, decodedValue);
+        } else {
+          throw new IllegalArgumentException(
+              "Invalid connection property format. Expected 'key=value', but got: " + pair);
+        }
+      }
+    } else {
+      try (StringReader reader = new StringReader(connectionProperties)) {
+        jdbcProperties.load(reader);
+      } catch (IOException e) {
+        LOG.error("Failed to parse connection properties", e);
+      }
+    }
+    return jdbcProperties;
+  }
 }

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java

@@ -33,6 +33,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.Properties;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -133,4 +134,90 @@ public void testInitConnectionPool() {
       }
     }
   }
+
+  @Test
+  public void testInitConnectionPoolWithUrlEncodedProperties() {
+    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");
+    // Test URL-encoded connection properties with & and URL-encoded characters
+    when(mockShard.getConnectionProperties())
+        .thenReturn("useSSL=true&requireSSL=true&encoded%26Key=encoded%3DValue");
+
+    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");
+    when(mockRequest.getJdbcUrlPrefix()).thenReturn("jdbc:mysql://");
+
+    try (MockedConstruction<HikariDataSource> mockedDsConstruction =
+        mockConstruction(
+            HikariDataSource.class,
+            (mock, context) -> when(mock.getConnection()).thenReturn(mock(Connection.class)))) {
+      try (MockedConstruction<HikariConfig> mockedConfigConstruction =
+          mockConstruction(HikariConfig.class)) {
+        connectionHelper.init(mockRequest);
+
+        assertTrue(connectionHelper.isConnectionPoolInitialized());
+
+        HikariConfig capturedConfig = mockedConfigConstruction.constructed().get(0);
+        // Verify both properties were split properly and encoded ones were decoded
+        verify(capturedConfig).addDataSourceProperty("useSSL", "true");
+        verify(capturedConfig).addDataSourceProperty("requireSSL", "true");
+        verify(capturedConfig).addDataSourceProperty("encoded&Key", "encoded=Value");
+        // Verify no other interactions
+      }
+    }
+  }
+
+  @Test
+  public void testParseProperties_nullOrEmpty() {
+    assertTrue(JdbcConnectionHelper.parseProperties(null).isEmpty());
+    assertTrue(JdbcConnectionHelper.parseProperties("").isEmpty());
+  }
+
+  @Test
+  public void testParseProperties_urlEncoded() {
+    String propsStr = "useSSL=true&requireSSL=true&encoded%26Key=encoded%3DValue";
+    Properties props = JdbcConnectionHelper.parseProperties(propsStr);
+
+    assertEquals(3, props.size());
+    assertEquals("true", props.getProperty("useSSL"));
+    assertEquals("true", props.getProperty("requireSSL"));
+    assertEquals("encoded=Value", props.getProperty("encoded&Key"));
+  }
+
+  @Test
+  public void testParseProperties_semicolonSeparated() {
+    String propsStr = "useSSL=true;requireSSL=true;encoded%26Key=encoded%3DValue";
+    Properties props = JdbcConnectionHelper.parseProperties(propsStr);
+
+    assertEquals(3, props.size());
+    assertEquals("true", props.getProperty("useSSL"));
+    assertEquals("true", props.getProperty("requireSSL"));
+    assertEquals("encoded=Value", props.getProperty("encoded&Key"));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testParseProperties_malformed() {
+    String propsStr = "useSSL=true&malformedParam";
+    JdbcConnectionHelper.parseProperties(propsStr);
+  }
+
+  @Test
+  public void testParseProperties_newlineSeparated() {
+    String propsStr = "useSSL=true\nrequireSSL=true\nencoded%26Key=encoded%3DValue";
+    Properties props = JdbcConnectionHelper.parseProperties(propsStr);
+
+    assertEquals(3, props.size());
+    assertEquals("true", props.getProperty("useSSL"));
+    assertEquals("true", props.getProperty("requireSSL"));
+    // Newline properties aren't URL decoded by Properties.load()
+    assertEquals("encoded%3DValue", props.getProperty("encoded%26Key"));
+  }
 }

v2/spanner-to-sourcedb/README.md

@@ -155,7 +155,8 @@ The file should be a list of JSONs as:
        "user": "root",
        "secretManagerUri": "projects/123/secrets/rev-cmek-cred-shard1/versions/latest",
        "port": "3306",
-       "dbName": "db1"
+       "dbName": "db1",
+       "connectionProperties": "useSSL=true&requireSSL=true"
      },
      {
        "logicalShardId": "shard2",
@@ -169,6 +170,15 @@ The file should be a list of JSONs as:
 }
 ```
 
+You can optionally specify `"connectionProperties"` to configure the JDBC connection (e.g. for SSL). The properties can be separated by `&` or `;` and the keys/values can be URL-encoded if they contain special characters.
+
+#### Referencing SSL Certificates
+To connect to a source database using a custom SSL certificate (e.g. a truststore), you must first make the certificate file available to the Dataflow workers:
+1. Upload your certificate file (e.g., `truststore.jks`) to a Google Cloud Storage bucket.
+2. When launching the Dataflow template, provide the GCS path to the `--extraFilesToStage` parameter (e.g. `--extraFilesToStage="gs://<BUCKET_NAME>/truststore.jks"`). The file will be downloaded to the `/extra_files` directory on each worker.
+3. In your shards JSON configuration, set the `"connectionProperties"` to reference this local file path. For example, for MySQL you would specify the `trustCertificateKeyStoreUrl` and password:
+   `"connectionProperties": "useSSL=true&requireSSL=true&trustCertificateKeyStoreUrl=file:/extra_files/truststore.jks&trustCertificateKeyStorePassword=my_password"`
+
 
 ### Sample source file for Cassandra
 

v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java

@@ -37,6 +37,7 @@
 import java.sql.ResultSet;
 import java.sql.Statement;
 import java.util.List;
+import java.util.Properties;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -142,6 +143,19 @@ Connection createConnection(Shard shard) throws Exception {
     config.setUsername(shard.getUserName());
     config.setPassword(shard.getPassword());
     config.setDriverClassName("com.mysql.cj.jdbc.Driver");
+
+    if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) {
+      LOG.info(
+          "Connection properties for shard {}: {}",
+          shard.getLogicalShardId(),
+          shard.getConnectionProperties());
+      Properties parsedProps =
+          JdbcConnectionHelper.parseProperties(shard.getConnectionProperties());
+      for (String key : parsedProps.stringPropertyNames()) {
+        config.addDataSourceProperty(key, parsedProps.getProperty(key));
+      }
+    }
+
     HikariDataSource ds = new HikariDataSource(config);
     return ds.getConnection();
   }

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java

@@ -21,6 +21,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
@@ -32,13 +33,17 @@
 import com.google.cloud.teleport.v2.templates.dbutils.dao.source.IDao;
 import com.google.cloud.teleport.v2.templates.dbutils.dao.source.JdbcDao;
 import com.google.cloud.teleport.v2.templates.dbutils.dml.IDMLGenerator;
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import java.sql.Connection;
 import java.util.Collections;
 import java.util.List;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mock;
+import org.mockito.MockedConstruction;
 import org.mockito.junit.MockitoJUnitRunner;
 
 @RunWith(MockitoJUnitRunner.class)
@@ -224,7 +229,7 @@ public void testGetInformationSchema() throws Exception {
     try (org.mockito.MockedConstruction<
             com.google.cloud.teleport.v2.spanner.sourceddl.MySqlInformationSchemaScanner>
         mocked =
-            org.mockito.Mockito.mockConstruction(
+            mockConstruction(
                 com.google.cloud.teleport.v2.spanner.sourceddl.MySqlInformationSchemaScanner.class,
                 (mock, context) -> {
                   when(mock.scan()).thenReturn(dummySchema);
@@ -275,4 +280,48 @@ public void testSupportsSharding() {
   public void testShouldUpdateReadValuesToSpannerRecord() {
     assertTrue(connector.shouldUpdateReadValuesToSpannerRecord());
   }
+
+  @Test
+  public void testCreateConnectionWithUrlEncodedProperties() throws Exception {
+    when(mockShard.getHost()).thenReturn("localhost");
+    when(mockShard.getPort()).thenReturn("3306");
+    when(mockShard.getDbName()).thenReturn("mydb");
+    when(mockShard.getUserName()).thenReturn("user");
+    when(mockShard.getPassword()).thenReturn("pass");
+    // Test URL-encoded connection properties with ; and URL-encoded characters
+    when(mockShard.getConnectionProperties())
+        .thenReturn("useSSL=true;requireSSL=true;encoded%26Key=encoded%3DValue");
+
+    try (MockedConstruction<HikariDataSource> mockedDsConstruction =
+        mockConstruction(
+            HikariDataSource.class,
+            (mock, context) -> {
+              when(mock.getConnection()).thenReturn(mock(Connection.class));
+            })) {
+      try (MockedConstruction<HikariConfig> mockedConfigConstruction =
+          mockConstruction(HikariConfig.class)) {
+
+        Connection conn = connector.createConnection(mockShard);
+        assertNotNull(conn);
+
+        HikariConfig capturedConfig = mockedConfigConstruction.constructed().get(0);
+        verify(capturedConfig).setJdbcUrl("jdbc:mysql://localhost:3306/mydb");
+        verify(capturedConfig).addDataSourceProperty("useSSL", "true");
+        verify(capturedConfig).addDataSourceProperty("requireSSL", "true");
+        verify(capturedConfig).addDataSourceProperty("encoded&Key", "encoded=Value");
+      }
+    }
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testCreateConnectionWithMalformedProperties() throws Exception {
+    when(mockShard.getHost()).thenReturn("localhost");
+    when(mockShard.getPort()).thenReturn("3306");
+    when(mockShard.getDbName()).thenReturn("mydb");
+    when(mockShard.getUserName()).thenReturn("user");
+    when(mockShard.getPassword()).thenReturn("pass");
+    when(mockShard.getConnectionProperties()).thenReturn("useSSL=true;malformedParam");
+
+    connector.createConnection(mockShard);
+  }
 }