From df7fa49e08867ab54b204444453a0b2574df3932 Mon Sep 17 00:00:00 2001 From: Maksim Ryzhukhin Date: Fri, 7 Aug 2026 01:08:47 +0200 Subject: [PATCH 1/2] [fix](cdc) apply SSL and JDBC config before startup mode resolves binlog offset The generateMySqlConfig() method applied SSL settings (ssl_mode, ssl_rootcert) and JDBC connection properties AFTER the startup mode switch-case block. Modes like "latest", "earliest", and 13-digit timestamp call initializeEffectiveOffset() inside that block, which opens a JDBC connection to resolve the current binlog position via SHOW MASTER STATUS. Since SSL was not yet configured on the factory, this connection used plaintext and failed on MySQL/Aurora instances with require_secure_transport=ON. Fix: move the JDBC properties + SSL + Debezium properties block to before the startup mode block, so initializeEffectiveOffset() uses the correct SSL settings when building its temporary connection. Modes "initial" and "snapshot" never call initializeEffectiveOffset() and are unaffected by this reordering. --- .../reader/mysql/MySqlSourceReader.java | 96 ++++++++++--------- 1 file changed, 50 insertions(+), 46 deletions(-) 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( From b7edfec7734a2899c9e8421ba562f329a9371520 Mon Sep 17 00:00:00 2001 From: Maksim Ryzhukhin Date: Sun, 9 Aug 2026 00:45:33 +0200 Subject: [PATCH 2/2] [test](cdc) add SSL regression test for startup-mode offset resolution Adds MySqlStartupSslITCase, which guards the ordering fixed in this PR. The test starts a MySQL container, enables require_secure_transport=ON after startup (so the Testcontainers readiness probe, which connects without TLS, is not broken), and runs a CDC job with offset=latest and ssl_mode=require. That offset mode reaches initializeEffectiveOffset(), the call that opened an un-encrypted JDBC connection before this fix. Test methods: - latestStartupModeSucceedsWithSecureTransport: the regression guard. Also asserts offset=latest does not replay pre-existing rows. - nonSslPathStillWorksWithoutSecureTransport: control case proving the reorder does not change the non-TLS path. Uses a SECOND container without the secure-transport requirement, rather than toggling the flag on a shared container, so a mid-test failure cannot leave it OFF and make the SSL assertions vacuous. - plaintextConnectionRejectedBySecureTransportServer: sanity check that the server really enforces TLS. All SQL the test issues against the TLS-enforcing container goes through the mysql CLI via execInContainer, because Unix socket connections are exempt from require_secure_transport while the test's own JDBC connections would otherwise be rejected too. Only offset=latest is covered. earliest and the timestamp mode go through the same initializeEffectiveOffset() path so they add no coverage of the ordering bug, and they need different assertions because they replay the setup rows instead of skipping them. Adds CdcClientWriteHarness.withSslMode() following the existing fluent withXxx() pattern. --- .../itcase/CdcClientWriteHarness.java | 6 + .../itcase/MySqlStartupSslITCase.java | 281 ++++++++++++++++++ 2 files changed, 287 insertions(+) create mode 100644 fs_brokers/cdc_client/src/test/java/org/apache/doris/cdcclient/itcase/MySqlStartupSslITCase.java 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); + } +}