diff --git a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java index 0ad2629ce94db6..e443020ec13efe 100644 --- a/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java +++ b/fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/mysql/MySqlSourceReader.java @@ -942,6 +942,56 @@ private MySqlSourceConfig generateMySqlConfig( tableList.length >= 1, "include_tables or table is required"); configFactory.tableList(tableList); + // JDBC properties and SSL must be configured BEFORE the startup mode block because + // modes like 'latest', 'earliest', and timestamp call initializeEffectiveOffset() + // which opens a JDBC connection to resolve the current binlog position. Without SSL + // configured here, that connection fails on servers with require_secure_transport=ON. + Properties jdbcProperteis = new Properties(); + jdbcProperteis.putAll(cu.getOriginalProperties()); + + Properties dbzProps = ConfigUtil.getDefaultDebeziumProps(); + dbzProps.setProperty( + MySqlConnectorConfig.KEEP_ALIVE_INTERVAL_MS.name(), + DEBEZIUM_HEARTBEAT_INTERVAL_MS + ""); + dbzProps.setProperty( + EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT, + Boolean.toString(excludeHeartbeatFromEventCount())); + + if (cdcConfig.containsKey(DataSourceConfigKeys.SSL_MODE)) { + String normalized = + normalizeSslModeForMysql(cdcConfig.get(DataSourceConfigKeys.SSL_MODE)); + dbzProps.put("database.ssl.mode", normalized); + // Flink CDC's forked MySqlConnection drops Debezium SSL props from the snapshot + // JDBC URL, so mirror to Connector/J native names. + jdbcProperteis.put("sslMode", normalized); + } + if (cdcConfig.containsKey(DataSourceConfigKeys.SSL_ROOTCERT)) { + String fileName = cdcConfig.get(DataSourceConfigKeys.SSL_ROOTCERT); + String truststorePath = SmallFileMgr.getPkcs12TruststorePath(fileName); + LOG.info("Using SSL truststore file path: {}", truststorePath); + dbzProps.put("database.ssl.truststore", truststorePath); + dbzProps.put("database.ssl.truststore.password", SmallFileMgr.TRUSTSTORE_PASSWORD); + jdbcProperteis.put("trustCertificateKeyStoreUrl", "file:" + truststorePath); + // Connector/J defaults keystore type to JKS; we generate PKCS12. + jdbcProperteis.put("trustCertificateKeyStoreType", "PKCS12"); + jdbcProperteis.put( + "trustCertificateKeyStorePassword", SmallFileMgr.TRUSTSTORE_PASSWORD); + } + + // Keep genuinely ancient (<100) DATE/DATETIME years; MySQL already completes 2-digit years. + dbzProps.setProperty("enable.time.adjuster", "false"); + // The converter is valid only when snapshot JDBC exposes YEAR values as numbers. + if ("false" + .equalsIgnoreCase( + jdbcProperteis.getProperty(PropertyKey.yearIsDateType.getKeyName()))) { + dbzProps.setProperty("converters", "dorisYear"); + dbzProps.setProperty("dorisYear.type", MySqlYearConverter.class.getName()); + } + + configFactory.jdbcProperties(jdbcProperteis); + configFactory.debeziumProperties(dbzProps); + configFactory.heartbeatInterval(Duration.ofMillis(DEBEZIUM_HEARTBEAT_INTERVAL_MS)); + // setting startMode String startupMode = cdcConfig.get(DataSourceConfigKeys.OFFSET); if (DataSourceConfigKeys.OFFSET_INITIAL.equalsIgnoreCase(startupMode)) { @@ -993,52 +1043,6 @@ private MySqlSourceConfig generateMySqlConfig( throw new RuntimeException("Unknown offset " + startupMode); } - Properties jdbcProperteis = new Properties(); - jdbcProperteis.putAll(cu.getOriginalProperties()); - configFactory.jdbcProperties(jdbcProperteis); - - Properties dbzProps = ConfigUtil.getDefaultDebeziumProps(); - dbzProps.setProperty( - MySqlConnectorConfig.KEEP_ALIVE_INTERVAL_MS.name(), - DEBEZIUM_HEARTBEAT_INTERVAL_MS + ""); - dbzProps.setProperty( - EXCLUDE_HEARTBEAT_FROM_EVENT_COUNT, - Boolean.toString(excludeHeartbeatFromEventCount())); - - if (cdcConfig.containsKey(DataSourceConfigKeys.SSL_MODE)) { - String normalized = - normalizeSslModeForMysql(cdcConfig.get(DataSourceConfigKeys.SSL_MODE)); - dbzProps.put("database.ssl.mode", normalized); - // Flink CDC's forked MySqlConnection drops Debezium SSL props from the snapshot - // JDBC URL, so mirror to Connector/J native names. - jdbcProperteis.put("sslMode", normalized); - } - if (cdcConfig.containsKey(DataSourceConfigKeys.SSL_ROOTCERT)) { - String fileName = cdcConfig.get(DataSourceConfigKeys.SSL_ROOTCERT); - String truststorePath = SmallFileMgr.getPkcs12TruststorePath(fileName); - LOG.info("Using SSL truststore file path: {}", truststorePath); - dbzProps.put("database.ssl.truststore", truststorePath); - dbzProps.put("database.ssl.truststore.password", SmallFileMgr.TRUSTSTORE_PASSWORD); - jdbcProperteis.put("trustCertificateKeyStoreUrl", "file:" + truststorePath); - // Connector/J defaults keystore type to JKS; we generate PKCS12. - jdbcProperteis.put("trustCertificateKeyStoreType", "PKCS12"); - jdbcProperteis.put( - "trustCertificateKeyStorePassword", SmallFileMgr.TRUSTSTORE_PASSWORD); - } - - // Keep genuinely ancient (<100) DATE/DATETIME years; MySQL already completes 2-digit years. - dbzProps.setProperty("enable.time.adjuster", "false"); - // The converter is valid only when snapshot JDBC exposes YEAR values as numbers. - if ("false" - .equalsIgnoreCase( - jdbcProperteis.getProperty(PropertyKey.yearIsDateType.getKeyName()))) { - dbzProps.setProperty("converters", "dorisYear"); - dbzProps.setProperty("dorisYear.type", MySqlYearConverter.class.getName()); - } - - configFactory.debeziumProperties(dbzProps); - configFactory.heartbeatInterval(Duration.ofMillis(DEBEZIUM_HEARTBEAT_INTERVAL_MS)); - configFactory.splitSize( Integer.parseInt( cdcConfig.getOrDefault( diff --git a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java index 07e5b40a811740..220db2b753aa87 100644 --- a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java +++ b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/CdcClientWriteHarness.java @@ -213,6 +213,12 @@ CdcClientWriteHarness withServerTimezone(String zoneId) { return this; } + /** Set the SSL mode used by the CDC connector (e.g. "require", "disable", "verify-ca"). */ + CdcClientWriteHarness withSslMode(String sslMode) { + config.put(DataSourceConfigKeys.SSL_MODE, sslMode); + return this; + } + private JobBaseConfig baseConfig() { return new JobBaseConfig(jobId, dataSource, config, null); } diff --git a/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/MySqlStartupSslITCase.java b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/MySqlStartupSslITCase.java new file mode 100644 index 00000000000000..6a67dc22a427b9 --- /dev/null +++ b/fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/MySqlStartupSslITCase.java @@ -0,0 +1,281 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.doris.cdcclient.itcase; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import org.apache.doris.cdcclient.common.Env; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.MySQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.utility.DockerImageName; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.SQLException; +import java.sql.Statement; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Regression test for PR #66559 — CDC SSL connection ordering fix. + * + *

Before the fix, {@code initializeEffectiveOffset()} was called before the SSL + * properties were applied to the Debezium JDBC connection, causing a connection failure on MySQL + * servers that enforce {@code require_secure_transport=ON}. The fix reorders SSL configuration + * to happen before the offset resolution JDBC call. + * + *

This test guards the correct ordering by: + *

+ */ +@Testcontainers +class MySqlStartupSslITCase { + + private static final String ROOT_USER = "root"; + private static final String ROOT_PASSWORD = "123456"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + private static final AtomicLong JOB_ID_SEQ = new AtomicLong(990_000); + + /** Container that will enforce TLS after startup (see {@link #enableSecureTransport}). */ + @Container + static final MySQLContainer MYSQL_SSL = + new MySQLContainer<>(DockerImageName.parse("mysql:8.0")) + .withDatabaseName("cdc_test") + .withUsername("cdc") + .withPassword("123456") + .withEnv("MYSQL_ROOT_PASSWORD", ROOT_PASSWORD); + + /** Separate container WITHOUT TLS enforcement — used for the non-SSL control test (D6). */ + @Container + static final MySQLContainer MYSQL_PLAIN = + new MySQLContainer<>(DockerImageName.parse("mysql:8.0")) + .withDatabaseName("cdc_test") + .withUsername("cdc") + .withPassword("123456") + .withEnv("MYSQL_ROOT_PASSWORD", ROOT_PASSWORD); + + private String jobId; + private String database; + + /** + * Enable {@code require_secure_transport=ON} AFTER container startup so the Testcontainers + * readiness probe (which connects without TLS) is not broken. + */ + @BeforeAll + static void enableSecureTransport() throws Exception { + // Use execInContainer with the mysql CLI so we do not need a TLS JDBC connection yet. + MYSQL_SSL.execInContainer( + "mysql", + "-uroot", + "-p" + ROOT_PASSWORD, + "-e", + "SET GLOBAL require_secure_transport=ON;"); + } + + @BeforeEach + void setUp() throws Exception { + jobId = String.valueOf(JOB_ID_SEQ.incrementAndGet()); + database = "ssl_db_" + jobId; + // Use execInContainer for DDL because the server now requires TLS. + execSql(MYSQL_SSL, + "CREATE DATABASE " + database + ";", + "USE " + database + ";", + "CREATE TABLE t_user (id INT PRIMARY KEY, name VARCHAR(50));", + "INSERT INTO t_user VALUES (1,'alice'), (2,'bob');"); + } + + @AfterEach + void tearDown() throws Exception { + Env.getCurrentEnv().close(jobId); + execSql(MYSQL_SSL, "DROP DATABASE IF EXISTS " + database + ";"); + } + + // ------------------------------------------------------------------------- + // offset=latest reaches initializeEffectiveOffset(), which is the call that + // opened the un-encrypted JDBC connection before PR #66559. + // + // Only 'latest' is covered here. 'earliest' and the timestamp mode go through + // the same initializeEffectiveOffset() path, so they add no extra coverage of + // the ordering bug, and they need different assertions because they replay the + // rows written during setUp() instead of skipping them. + // ------------------------------------------------------------------------- + + @Test + void latestStartupModeSucceedsWithSecureTransport() throws Exception { + try (MockDorisServer mock = new MockDorisServer(); + CdcClientWriteHarness harness = + CdcClientWriteHarness.mysql( + jobId, + MYSQL_SSL.getHost(), + MYSQL_SSL.getMappedPort(MySQLContainer.MYSQL_PORT), + ROOT_USER, + ROOT_PASSWORD, + database, + "t_user", + "latest", + "doris_target_db", + mock) + .withSslMode("require")) { + + // Resolves the latest binlog position over TLS. Before PR #66559 this threw, + // because the SSL properties were attached to the config factory only after + // this call had already opened its JDBC connection. + harness.enterBinlogFromStartupMode(); + + execSql(MYSQL_SSL, + "USE " + database + ";", + "INSERT INTO t_user VALUES (3,'carol');"); + + List streamed = ids(harness.continueBinlog(1, Duration.ofSeconds(90))); + assertThat(streamed).containsExactly(3); + + // offset=latest must not replay the rows that existed before the job started. + assertThat(ids(harness.loadedRecords())).doesNotContain(1, 2); + } + } + + // ------------------------------------------------------------------------- + // Control: non-TLS path still works (uses separate MYSQL_PLAIN container — D6 fix) + // ------------------------------------------------------------------------- + + @Test + void nonSslPathStillWorksWithoutSecureTransport() throws Exception { + String plainDb = "plain_db_" + jobId; + try (Connection conn = plainRootConnection(""); + Statement st = conn.createStatement()) { + st.execute("CREATE DATABASE " + plainDb); + st.execute("USE " + plainDb); + st.execute("CREATE TABLE t_user (id INT PRIMARY KEY, name VARCHAR(50))"); + st.execute("INSERT INTO t_user VALUES (1,'alice')"); + } + + try (MockDorisServer mock = new MockDorisServer(); + CdcClientWriteHarness harness = + CdcClientWriteHarness.mysql( + jobId, + MYSQL_PLAIN.getHost(), + MYSQL_PLAIN.getMappedPort(MySQLContainer.MYSQL_PORT), + ROOT_USER, + ROOT_PASSWORD, + plainDb, + "t_user", + "latest", + "doris_target_db", + mock)) { + + // No .withSslMode() — default (no TLS) path must still work. + harness.enterBinlogFromStartupMode(); + + try (Connection conn = plainRootConnection(plainDb); + Statement st = conn.createStatement()) { + st.execute("INSERT INTO t_user VALUES (2,'bob')"); + } + + List rows = harness.continueBinlog(1, Duration.ofSeconds(90)); + assertThat(ids(rows)).containsExactly(2); + } finally { + try (Connection conn = plainRootConnection(""); + Statement st = conn.createStatement()) { + st.execute("DROP DATABASE IF EXISTS " + plainDb); + } + } + } + + // ------------------------------------------------------------------------- + // Sanity: plaintext connection IS rejected by the TLS-enforcing container + // ------------------------------------------------------------------------- + + @Test + void plaintextConnectionRejectedBySecureTransportServer() { + // Attempt a plaintext JDBC connection to the TLS-enforcing container. + String url = "jdbc:mysql://" + + MYSQL_SSL.getHost() + + ":" + + MYSQL_SSL.getMappedPort(MySQLContainer.MYSQL_PORT) + + "/cdc_test" + + "?sslMode=DISABLED&allowPublicKeyRetrieval=true"; + assertThatThrownBy(() -> DriverManager.getConnection(url, ROOT_USER, ROOT_PASSWORD)) + .isInstanceOf(SQLException.class) + .hasMessageContaining("secure transport"); + } + + // ------------------------------------------------------------------------- + // Helpers + // ------------------------------------------------------------------------- + + /** + * Execute SQL statements via {@code execInContainer} using the mysql CLI. This avoids needing + * a TLS-capable JDBC connection for the SSL-enforcing container. + */ + private static void execSql(MySQLContainer container, String... statements) + throws Exception { + StringBuilder sb = new StringBuilder(); + for (String stmt : statements) { + sb.append(stmt); + if (!stmt.endsWith(";")) { + sb.append(";"); + } + } + org.testcontainers.containers.Container.ExecResult result = container.execInContainer( + "mysql", + "-uroot", + "-p" + ROOT_PASSWORD, + "-e", + sb.toString()); + if (result.getExitCode() != 0) { + throw new RuntimeException( + "mysql exec failed (exit " + result.getExitCode() + "): " + result.getStderr()); + } + } + + private List ids(List records) throws Exception { + List result = new ArrayList<>(); + for (String record : records) { + JsonNode node = MAPPER.readTree(record); + result.add(node.get("id").asInt()); + } + return result; + } + + /** Plain JDBC connection to the non-TLS container (MYSQL_PLAIN). */ + private Connection plainRootConnection(String db) throws Exception { + String url = "jdbc:mysql://" + + MYSQL_PLAIN.getHost() + + ":" + + MYSQL_PLAIN.getMappedPort(MySQLContainer.MYSQL_PORT) + + "/" + + db; + return DriverManager.getConnection(url, ROOT_USER, ROOT_PASSWORD); + } +}