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-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..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 @@ -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 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) { + 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; + } } 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..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 @@ -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 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 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")); + } } 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 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..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 @@ -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(); } 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..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 @@ -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,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 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"); + 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); + } }