From 79279d8ee532e23697a768125b6fc9d294bddf6b Mon Sep 17 00:00:00 2001 From: Darshan Siddesh Jagaluru Date: Thu, 27 Aug 2026 12:47:37 +0000 Subject: [PATCH 1/7] [SpannerToSourceDb] Add support for URL-encoded connection properties to enable MySQL SSL in spanner-to-sourcedb --- .../connection/JdbcConnectionHelper.java | 19 ++++++++-- .../connection/JdbcConnectionHelperTest.java | 38 +++++++++++++++++++ .../mysql/MySQLSpToSrcSourceConnector.java | 24 ++++++++++++ .../MySQLSpToSrcSourceConnectorTest.java | 37 +++++++++++++++++- 4 files changed, 113 insertions(+), 5 deletions(-) diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java index 4dca05c070..a32a1bb65a 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java @@ -71,10 +71,21 @@ 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()); + String props = shard.getConnectionProperties(); + if (props.contains("&") || props.contains(";")) { + String[] pairs = props.split("[&;]"); + for (String pair : pairs) { + String[] kv = pair.split("=", 2); + if (kv.length == 2) { + jdbcProperties.setProperty(kv[0], kv[1]); + } + } + } else { + try (StringReader reader = new StringReader(props)) { + jdbcProperties.load(reader); + } catch (IOException e) { + LOG.error("Error converting string to properties: {}", e.getMessage()); + } } } diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java index 3b0f3e88d9..d322c3a77f 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java +++ b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java @@ -133,4 +133,42 @@ 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 a malformed parameter to cover kv.length != 2 branch + when(mockShard.getConnectionProperties()).thenReturn("useSSL=true&requireSSL=true&malformedParam"); + + List 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 mockedDsConstruction = + mockConstruction( + HikariDataSource.class, + (mock, context) -> when(mock.getConnection()).thenReturn(mock(Connection.class)))) { + try (MockedConstruction mockedConfigConstruction = + mockConstruction(HikariConfig.class)) { + connectionHelper.init(mockRequest); + + assertTrue(connectionHelper.isConnectionPoolInitialized()); + + HikariConfig capturedConfig = mockedConfigConstruction.constructed().get(0); + // Verify both properties were split properly and the malformed one was ignored + verify(capturedConfig).addDataSourceProperty("useSSL", "true"); + verify(capturedConfig).addDataSourceProperty("requireSSL", "true"); + // Verify no other interactions (meaning malformedParam wasn't added) + } + } + } } diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java index ac4043f135..e33d9338a5 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java @@ -33,10 +33,12 @@ import com.google.common.annotations.VisibleForTesting; import com.zaxxer.hikari.HikariConfig; import com.zaxxer.hikari.HikariDataSource; +import java.io.StringReader; import java.sql.Connection; 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 +144,28 @@ 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()) { + String props = shard.getConnectionProperties(); + if (props.contains("&") || props.contains(";")) { + String[] pairs = props.split("[&;]"); + for (String pair : pairs) { + String[] kv = pair.split("=", 2); + if (kv.length == 2) { + config.addDataSourceProperty(kv[0], kv[1]); + } + } + } else { + Properties jdbcProperties = new Properties(); + try (StringReader reader = new StringReader(props)) { + jdbcProperties.load(reader); + for (String key : jdbcProperties.stringPropertyNames()) { + config.addDataSourceProperty(key, jdbcProperties.getProperty(key)); + } + } + } + } + HikariDataSource ds = new HikariDataSource(config); return ds.getConnection(); } diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java index 87fc6ed57d..98fc7a742f 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java +++ b/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,6 +33,9 @@ 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; @@ -39,6 +43,7 @@ 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,34 @@ 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 a malformed parameter to cover kv.length != 2 branch + when(mockShard.getConnectionProperties()).thenReturn("useSSL=true;requireSSL=true;malformedParam"); + + try (MockedConstruction mockedDsConstruction = + mockConstruction( + HikariDataSource.class, + (mock, context) -> { + when(mock.getConnection()).thenReturn(mock(Connection.class)); + })) { + try (MockedConstruction 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"); + } + } + } } From 0d5bd739143268c77d1a9f9760aa62163aaf7b2f Mon Sep 17 00:00:00 2001 From: Darshan Siddesh Jagaluru Date: Thu, 27 Aug 2026 12:59:42 +0000 Subject: [PATCH 2/7] Formatting changes --- .../migrations/connection/JdbcConnectionHelperTest.java | 8 +++++--- .../source/mysql/MySQLSpToSrcSourceConnector.java | 2 +- .../source/mysql/MySQLSpToSrcSourceConnectorTest.java | 6 ++++-- 3 files changed, 10 insertions(+), 6 deletions(-) diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java index d322c3a77f..4332e72394 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java +++ b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java @@ -143,8 +143,10 @@ public void testInitConnectionPoolWithUrlEncodedProperties() { when(mockShard.getDbName()).thenReturn("testdb"); when(mockShard.getUserName()).thenReturn("testuser"); when(mockShard.getPassword()).thenReturn("testpassword"); - // Test URL-encoded connection properties with & and a malformed parameter to cover kv.length != 2 branch - when(mockShard.getConnectionProperties()).thenReturn("useSSL=true&requireSSL=true&malformedParam"); + // Test URL-encoded connection properties with & and a malformed parameter to cover kv.length != + // 2 branch + when(mockShard.getConnectionProperties()) + .thenReturn("useSSL=true&requireSSL=true&malformedParam"); List mockShards = Collections.singletonList(mockShard); when(mockRequest.getShards()).thenReturn(mockShards); @@ -162,7 +164,7 @@ public void testInitConnectionPoolWithUrlEncodedProperties() { connectionHelper.init(mockRequest); assertTrue(connectionHelper.isConnectionPoolInitialized()); - + HikariConfig capturedConfig = mockedConfigConstruction.constructed().get(0); // Verify both properties were split properly and the malformed one was ignored verify(capturedConfig).addDataSourceProperty("useSSL", "true"); diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java index e33d9338a5..0e34104b87 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java @@ -144,7 +144,7 @@ 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()) { String props = shard.getConnectionProperties(); if (props.contains("&") || props.contains(";")) { diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java index 98fc7a742f..0f8939a204 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java @@ -288,8 +288,10 @@ public void testCreateConnectionWithUrlEncodedProperties() throws Exception { when(mockShard.getDbName()).thenReturn("mydb"); when(mockShard.getUserName()).thenReturn("user"); when(mockShard.getPassword()).thenReturn("pass"); - // Test URL-encoded connection properties with ; and a malformed parameter to cover kv.length != 2 branch - when(mockShard.getConnectionProperties()).thenReturn("useSSL=true;requireSSL=true;malformedParam"); + // Test URL-encoded connection properties with ; and a malformed parameter to cover kv.length != + // 2 branch + when(mockShard.getConnectionProperties()) + .thenReturn("useSSL=true;requireSSL=true;malformedParam"); try (MockedConstruction mockedDsConstruction = mockConstruction( From afb7690e59457413edeffd37197037c08e36cbe1 Mon Sep 17 00:00:00 2001 From: Darshan Siddesh Jagaluru Date: Thu, 27 Aug 2026 13:28:57 +0000 Subject: [PATCH 3/7] Addressing review comments --- .../migrations/connection/JdbcConnectionHelper.java | 6 +++++- .../connection/JdbcConnectionHelperTest.java | 10 ++++++---- .../source/mysql/MySQLSpToSrcSourceConnector.java | 6 +++++- .../source/mysql/MySQLSpToSrcSourceConnectorTest.java | 5 +++-- 4 files changed, 19 insertions(+), 8 deletions(-) diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java index a32a1bb65a..557da89f0c 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java @@ -21,6 +21,8 @@ import com.zaxxer.hikari.HikariDataSource; import java.io.IOException; import java.io.StringReader; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.util.HashMap; import java.util.Map; @@ -77,7 +79,9 @@ public synchronized void init(ConnectionHelperRequest connectionHelperRequest) { for (String pair : pairs) { String[] kv = pair.split("=", 2); if (kv.length == 2) { - jdbcProperties.setProperty(kv[0], kv[1]); + String decodedKey = URLDecoder.decode(kv[0], StandardCharsets.UTF_8); + String decodedValue = URLDecoder.decode(kv[1], StandardCharsets.UTF_8); + jdbcProperties.setProperty(decodedKey, decodedValue); } } } else { diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java index 4332e72394..3196948d6f 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java +++ b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java @@ -143,10 +143,10 @@ public void testInitConnectionPoolWithUrlEncodedProperties() { when(mockShard.getDbName()).thenReturn("testdb"); when(mockShard.getUserName()).thenReturn("testuser"); when(mockShard.getPassword()).thenReturn("testpassword"); - // Test URL-encoded connection properties with & and a malformed parameter to cover kv.length != - // 2 branch + // Test URL-encoded connection properties with &, a malformed parameter to cover kv.length != 2 + // branch, and URL-encoded characters when(mockShard.getConnectionProperties()) - .thenReturn("useSSL=true&requireSSL=true&malformedParam"); + .thenReturn("useSSL=true&requireSSL=true&malformedParam&encoded%26Key=encoded%3DValue"); List mockShards = Collections.singletonList(mockShard); when(mockRequest.getShards()).thenReturn(mockShards); @@ -166,9 +166,11 @@ public void testInitConnectionPoolWithUrlEncodedProperties() { assertTrue(connectionHelper.isConnectionPoolInitialized()); HikariConfig capturedConfig = mockedConfigConstruction.constructed().get(0); - // Verify both properties were split properly and the malformed one was ignored + // Verify both properties were split properly, malformed one was ignored, 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 (meaning malformedParam wasn't added) } } diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java index 0e34104b87..9a25ea54bc 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java @@ -34,6 +34,8 @@ import com.zaxxer.hikari.HikariConfig; import com.zaxxer.hikari.HikariDataSource; import java.io.StringReader; +import java.net.URLDecoder; +import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.sql.ResultSet; import java.sql.Statement; @@ -152,7 +154,9 @@ Connection createConnection(Shard shard) throws Exception { for (String pair : pairs) { String[] kv = pair.split("=", 2); if (kv.length == 2) { - config.addDataSourceProperty(kv[0], kv[1]); + String decodedKey = URLDecoder.decode(kv[0], StandardCharsets.UTF_8); + String decodedValue = URLDecoder.decode(kv[1], StandardCharsets.UTF_8); + config.addDataSourceProperty(decodedKey, decodedValue); } } } else { diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java index 0f8939a204..709502876b 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java @@ -289,9 +289,9 @@ public void testCreateConnectionWithUrlEncodedProperties() throws Exception { when(mockShard.getUserName()).thenReturn("user"); when(mockShard.getPassword()).thenReturn("pass"); // Test URL-encoded connection properties with ; and a malformed parameter to cover kv.length != - // 2 branch + // 2 branch, and URL-encoded characters when(mockShard.getConnectionProperties()) - .thenReturn("useSSL=true;requireSSL=true;malformedParam"); + .thenReturn("useSSL=true;requireSSL=true;malformedParam;encoded%26Key=encoded%3DValue"); try (MockedConstruction mockedDsConstruction = mockConstruction( @@ -309,6 +309,7 @@ public void testCreateConnectionWithUrlEncodedProperties() throws Exception { 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"); } } } From 44ec8ddf7295e065e6de1a662c1d9efe4af5b12f Mon Sep 17 00:00:00 2001 From: Darshan Siddesh Jagaluru Date: Fri, 28 Aug 2026 05:56:08 +0000 Subject: [PATCH 4/7] Moved to util class --- .../connection/JdbcConnectionHelper.java | 63 ++++++++++++------- .../connection/JdbcConnectionHelperTest.java | 41 ++++++++++++ .../mysql/MySQLSpToSrcSourceConnector.java | 25 +------- 3 files changed, 86 insertions(+), 43 deletions(-) diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java index 557da89f0c..ed0d423d1b 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java @@ -19,10 +19,6 @@ import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; import com.zaxxer.hikari.HikariConfig; import com.zaxxer.hikari.HikariDataSource; -import java.io.IOException; -import java.io.StringReader; -import java.net.URLDecoder; -import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.util.HashMap; import java.util.Map; @@ -73,23 +69,9 @@ public synchronized void init(ConnectionHelperRequest connectionHelperRequest) { config.setMinimumIdle(0); // avoid pre-filling connections Properties jdbcProperties = new Properties(); if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) { - String props = shard.getConnectionProperties(); - if (props.contains("&") || props.contains(";")) { - String[] pairs = props.split("[&;]"); - for (String pair : pairs) { - String[] kv = pair.split("=", 2); - if (kv.length == 2) { - String decodedKey = URLDecoder.decode(kv[0], StandardCharsets.UTF_8); - String decodedValue = URLDecoder.decode(kv[1], StandardCharsets.UTF_8); - jdbcProperties.setProperty(decodedKey, decodedValue); - } - } - } else { - try (StringReader reader = new StringReader(props)) { - jdbcProperties.load(reader); - } catch (IOException e) { - LOG.error("Error converting string to properties: {}", e.getMessage()); - } + Properties parsedProps = parseProperties(shard.getConnectionProperties()); + for (String key : parsedProps.stringPropertyNames()) { + jdbcProperties.setProperty(key, parsedProps.getProperty(key)); } } @@ -126,4 +108,43 @@ public Connection getConnection(String connectionRequestKey) throws ConnectionEx public void setConnectionPoolMap(Map inputMap) { connectionPoolMap = inputMap; } + + /** + * Parses connection properties from a string into a {@link Properties} object. + * + *

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) { + try { + String decodedKey = java.net.URLDecoder.decode(kv[0], java.nio.charset.StandardCharsets.UTF_8.name()); + String decodedValue = java.net.URLDecoder.decode(kv[1], java.nio.charset.StandardCharsets.UTF_8.name()); + jdbcProperties.setProperty(decodedKey, decodedValue); + } catch (java.io.UnsupportedEncodingException e) { + LOG.error("UTF-8 encoding not supported", e); + } + } + } + } else { + try (java.io.StringReader reader = new java.io.StringReader(connectionProperties)) { + jdbcProperties.load(reader); + } catch (java.io.IOException e) { + LOG.error("Failed to parse connection properties", e); + } + } + return jdbcProperties; + } } diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java index 3196948d6f..4c643031ee 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java +++ b/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; @@ -175,4 +176,44 @@ public void testInitConnectionPoolWithUrlEncodedProperties() { } } } + + @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&malformedParam&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;malformedParam;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_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")); + } } diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java index 9a25ea54bc..1238321f88 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java @@ -33,9 +33,6 @@ import com.google.common.annotations.VisibleForTesting; import com.zaxxer.hikari.HikariConfig; import com.zaxxer.hikari.HikariDataSource; -import java.io.StringReader; -import java.net.URLDecoder; -import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.sql.ResultSet; import java.sql.Statement; @@ -148,25 +145,9 @@ Connection createConnection(Shard shard) throws Exception { config.setDriverClassName("com.mysql.cj.jdbc.Driver"); if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) { - String props = shard.getConnectionProperties(); - if (props.contains("&") || props.contains(";")) { - String[] pairs = props.split("[&;]"); - for (String pair : pairs) { - String[] kv = pair.split("=", 2); - if (kv.length == 2) { - String decodedKey = URLDecoder.decode(kv[0], StandardCharsets.UTF_8); - String decodedValue = URLDecoder.decode(kv[1], StandardCharsets.UTF_8); - config.addDataSourceProperty(decodedKey, decodedValue); - } - } - } else { - Properties jdbcProperties = new Properties(); - try (StringReader reader = new StringReader(props)) { - jdbcProperties.load(reader); - for (String key : jdbcProperties.stringPropertyNames()) { - config.addDataSourceProperty(key, jdbcProperties.getProperty(key)); - } - } + Properties parsedProps = JdbcConnectionHelper.parseProperties(shard.getConnectionProperties()); + for (String key : parsedProps.stringPropertyNames()) { + config.addDataSourceProperty(key, parsedProps.getProperty(key)); } } From db632fe5e2c76caaf6a144ddbd23481cb07a408d Mon Sep 17 00:00:00 2001 From: Darshan Siddesh Jagaluru Date: Fri, 28 Aug 2026 08:34:39 +0000 Subject: [PATCH 5/7] Added log --- .../v2/spanner/migrations/connection/JdbcConnectionHelper.java | 2 ++ .../v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java | 2 ++ 2 files changed, 4 insertions(+) diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java index ed0d423d1b..da1f3131cc 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java @@ -70,6 +70,8 @@ public synchronized void init(ConnectionHelperRequest connectionHelperRequest) { Properties jdbcProperties = new Properties(); if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) { Properties parsedProps = parseProperties(shard.getConnectionProperties()); + LOG.info( + "Connection properties for shard {}: {}", shard.getLogicalShardId(), shard.getConnectionProperties()); for (String key : parsedProps.stringPropertyNames()) { jdbcProperties.setProperty(key, parsedProps.getProperty(key)); } diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java index 1238321f88..a4684b627e 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java @@ -146,6 +146,8 @@ Connection createConnection(Shard shard) throws Exception { if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) { Properties parsedProps = JdbcConnectionHelper.parseProperties(shard.getConnectionProperties()); + LOG.info( + "Connection properties for shard {}: {}", shard.getLogicalShardId(), shard.getConnectionProperties()); for (String key : parsedProps.stringPropertyNames()) { config.addDataSourceProperty(key, parsedProps.getProperty(key)); } From 886abf048775eb300a3fe1383a2bb00d8c750426 Mon Sep 17 00:00:00 2001 From: Darshan Siddesh Jagaluru Date: Fri, 28 Aug 2026 08:45:10 +0000 Subject: [PATCH 6/7] Added malformed parameter handling --- .../connection/JdbcConnectionHelper.java | 26 +++++++++++-------- .../connection/JdbcConnectionHelperTest.java | 20 ++++++++------ .../mysql/MySQLSpToSrcSourceConnector.java | 7 +++-- .../MySQLSpToSrcSourceConnectorTest.java | 17 +++++++++--- 4 files changed, 46 insertions(+), 24 deletions(-) diff --git a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java index da1f3131cc..9fa17a7b33 100644 --- a/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java +++ b/v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelper.java @@ -19,6 +19,9 @@ import com.google.cloud.teleport.v2.spanner.migrations.shard.Shard; import com.zaxxer.hikari.HikariConfig; 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; @@ -69,9 +72,11 @@ public synchronized void init(ConnectionHelperRequest connectionHelperRequest) { config.setMinimumIdle(0); // avoid pre-filling connections Properties jdbcProperties = new Properties(); if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) { - Properties parsedProps = parseProperties(shard.getConnectionProperties()); LOG.info( - "Connection properties for shard {}: {}", shard.getLogicalShardId(), shard.getConnectionProperties()); + "Connection properties for shard {}: {}", + shard.getLogicalShardId(), + shard.getConnectionProperties()); + Properties parsedProps = parseProperties(shard.getConnectionProperties()); for (String key : parsedProps.stringPropertyNames()) { jdbcProperties.setProperty(key, parsedProps.getProperty(key)); } @@ -131,19 +136,18 @@ public static Properties parseProperties(String connectionProperties) { for (String pair : pairs) { String[] kv = pair.split("=", 2); if (kv.length == 2) { - try { - String decodedKey = java.net.URLDecoder.decode(kv[0], java.nio.charset.StandardCharsets.UTF_8.name()); - String decodedValue = java.net.URLDecoder.decode(kv[1], java.nio.charset.StandardCharsets.UTF_8.name()); - jdbcProperties.setProperty(decodedKey, decodedValue); - } catch (java.io.UnsupportedEncodingException e) { - LOG.error("UTF-8 encoding not supported", e); - } + 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 (java.io.StringReader reader = new java.io.StringReader(connectionProperties)) { + try (StringReader reader = new StringReader(connectionProperties)) { jdbcProperties.load(reader); - } catch (java.io.IOException e) { + } catch (IOException e) { LOG.error("Failed to parse connection properties", e); } } diff --git a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java index 4c643031ee..e5d97a7c8b 100644 --- a/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java +++ b/v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/migrations/connection/JdbcConnectionHelperTest.java @@ -144,10 +144,9 @@ public void testInitConnectionPoolWithUrlEncodedProperties() { when(mockShard.getDbName()).thenReturn("testdb"); when(mockShard.getUserName()).thenReturn("testuser"); when(mockShard.getPassword()).thenReturn("testpassword"); - // Test URL-encoded connection properties with &, a malformed parameter to cover kv.length != 2 - // branch, and URL-encoded characters + // Test URL-encoded connection properties with & and URL-encoded characters when(mockShard.getConnectionProperties()) - .thenReturn("useSSL=true&requireSSL=true&malformedParam&encoded%26Key=encoded%3DValue"); + .thenReturn("useSSL=true&requireSSL=true&encoded%26Key=encoded%3DValue"); List mockShards = Collections.singletonList(mockShard); when(mockRequest.getShards()).thenReturn(mockShards); @@ -167,12 +166,11 @@ public void testInitConnectionPoolWithUrlEncodedProperties() { assertTrue(connectionHelper.isConnectionPoolInitialized()); HikariConfig capturedConfig = mockedConfigConstruction.constructed().get(0); - // Verify both properties were split properly, malformed one was ignored, and encoded ones - // were decoded + // 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 (meaning malformedParam wasn't added) + // Verify no other interactions } } } @@ -185,7 +183,7 @@ public void testParseProperties_nullOrEmpty() { @Test public void testParseProperties_urlEncoded() { - String propsStr = "useSSL=true&requireSSL=true&malformedParam&encoded%26Key=encoded%3DValue"; + String propsStr = "useSSL=true&requireSSL=true&encoded%26Key=encoded%3DValue"; Properties props = JdbcConnectionHelper.parseProperties(propsStr); assertEquals(3, props.size()); @@ -196,7 +194,7 @@ public void testParseProperties_urlEncoded() { @Test public void testParseProperties_semicolonSeparated() { - String propsStr = "useSSL=true;requireSSL=true;malformedParam;encoded%26Key=encoded%3DValue"; + String propsStr = "useSSL=true;requireSSL=true;encoded%26Key=encoded%3DValue"; Properties props = JdbcConnectionHelper.parseProperties(propsStr); assertEquals(3, props.size()); @@ -205,6 +203,12 @@ public void testParseProperties_semicolonSeparated() { 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"; diff --git a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java index a4684b627e..adcc123fec 100644 --- a/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java +++ b/v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnector.java @@ -145,9 +145,12 @@ Connection createConnection(Shard shard) throws Exception { config.setDriverClassName("com.mysql.cj.jdbc.Driver"); if (shard.getConnectionProperties() != null && !shard.getConnectionProperties().isEmpty()) { - Properties parsedProps = JdbcConnectionHelper.parseProperties(shard.getConnectionProperties()); LOG.info( - "Connection properties for shard {}: {}", shard.getLogicalShardId(), shard.getConnectionProperties()); + "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)); } diff --git a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java index 709502876b..7b1a87ecd8 100644 --- a/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java +++ b/v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/source/mysql/MySQLSpToSrcSourceConnectorTest.java @@ -288,10 +288,9 @@ public void testCreateConnectionWithUrlEncodedProperties() throws Exception { when(mockShard.getDbName()).thenReturn("mydb"); when(mockShard.getUserName()).thenReturn("user"); when(mockShard.getPassword()).thenReturn("pass"); - // Test URL-encoded connection properties with ; and a malformed parameter to cover kv.length != - // 2 branch, and URL-encoded characters + // Test URL-encoded connection properties with ; and URL-encoded characters when(mockShard.getConnectionProperties()) - .thenReturn("useSSL=true;requireSSL=true;malformedParam;encoded%26Key=encoded%3DValue"); + .thenReturn("useSSL=true;requireSSL=true;encoded%26Key=encoded%3DValue"); try (MockedConstruction mockedDsConstruction = mockConstruction( @@ -313,4 +312,16 @@ public void testCreateConnectionWithUrlEncodedProperties() throws Exception { } } } + + @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); + } } From 460ece5be2d2e64e40a764be51db18287acb7f10 Mon Sep 17 00:00:00 2001 From: Darshan Siddesh Jagaluru Date: Fri, 28 Aug 2026 09:02:17 +0000 Subject: [PATCH 7/7] Readme changes --- v2/sourcedb-to-spanner/README.md | 8 +++++++- v2/spanner-to-sourcedb/README.md | 12 +++++++++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/v2/sourcedb-to-spanner/README.md b/v2/sourcedb-to-spanner/README.md index bd66ed30d3..203f81d5db 100644 --- a/v2/sourcedb-to-spanner/README.md +++ b/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:///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). diff --git a/v2/spanner-to-sourcedb/README.md b/v2/spanner-to-sourcedb/README.md index eb5112b46c..d96b0dbaef 100644 --- a/v2/spanner-to-sourcedb/README.md +++ b/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:///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