Skip to content

[FLINK-38522][cdc connector mysql] use ssl for BinaryLogClient when searching for binlog offset for starting mysql cdc from timestamp #4156

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
+203 −0
Open

[FLINK-38522][cdc connector mysql] use ssl for BinaryLogClient when searching for binlog offset for starting mysql cdc from timestamp #4156

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
50cf4e8
[FLINK-38522][cdc connector mysql] use ssl for BinaryLogClient when s…
watsonjo737 Oct 14, 2025
111681c
[FLINK-38522][cdc connector mysql]: accommodate changes from PR review
watsonjo737 Oct 21, 2025
22e99b7
[FLINK-38522][cdc connector mysql]: accommodate changes from PR revie…
watsonjo737 Oct 22, 2025
8aedf5f
[FLINK-38522][cdc connector mysql]: add IT which shows how to configu…
watsonjo737 Oct 22, 2025
6615f91
[FLINK-38522][cdc connector mysql]: ensure new IT Test fails when SSL…
watsonjo737 Oct 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions ...mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import com.github.shyiko.mysql.binlog.event.EventData;
import com.github.shyiko.mysql.binlog.event.EventHeaderV4;
import com.github.shyiko.mysql.binlog.event.RotateEventData;
import com.github.shyiko.mysql.binlog.network.SSLMode;
import io.debezium.config.Configuration;
import io.debezium.connector.mysql.MySqlConnection;
import io.debezium.connector.mysql.MySqlConnectorConfig;
Expand Down Expand Up @@ -242,12 +243,23 @@ private static Map<String, String> querySystemVariables(
return variables;
}

static SSLMode sslModeFor(MySqlConnectorConfig.SecureConnectionMode mode) {
try {
return mode == null ? null : SSLMode.valueOf(mode.name());
} catch (IllegalArgumentException e) {
LOG.error("Invalid SecureConnectionMode provided %s", mode.name(), e);
}
return null;
}

public static BinlogOffset findBinlogOffset(
long targetMs, MySqlConnection connection, MySqlSourceConfig mySqlSourceConfig) {
MySqlConnection.MySqlConnectionConfiguration config = connection.connectionConfig();
BinaryLogClient client =
new BinaryLogClient(
config.hostname(), config.port(), config.username(), config.password());
client.setSSLMode(sslModeFor(config.sslMode()));

if (mySqlSourceConfig.getServerIdRange() != null) {
client.setServerId(mySqlSourceConfig.getServerIdRange().getStartServerId());
}
Expand Down
26 changes: 26 additions & 0 deletions ...l-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtilsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,13 +21,19 @@
import org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceConfigFactory;
import org.apache.flink.cdc.connectors.mysql.table.StartupOptions;

import com.github.shyiko.mysql.binlog.network.SSLMode;
import io.debezium.connector.mysql.MySqlConnection;
import io.debezium.connector.mysql.MySqlConnectorConfig;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;

import java.time.Duration;
import java.time.ZoneId;
import java.util.Properties;
import java.util.stream.Stream;

/** Tests for {@link org.apache.flink.cdc.connectors.mysql.debezium.DebeziumUtils}. */
class DebeziumUtilsTest {
Expand Down Expand Up @@ -83,6 +89,26 @@ private MySqlSourceConfig getConfig(Properties jdbcProperties) {
.createConfig(0);
}

@ParameterizedTest
@MethodSource("sslModeProvider")
void testSslModeConversion(MySqlConnectorConfig.SecureConnectionMode input, SSLMode expected) {
SSLMode actual = DebeziumUtils.sslModeFor(input);
Assertions.assertThat(actual).isEqualTo(expected);
}

static Stream<Arguments> sslModeProvider() {
return Stream.of(
Arguments.of(MySqlConnectorConfig.SecureConnectionMode.DISABLED, SSLMode.DISABLED),
Arguments.of(
MySqlConnectorConfig.SecureConnectionMode.PREFERRED, SSLMode.PREFERRED),
Arguments.of(MySqlConnectorConfig.SecureConnectionMode.REQUIRED, SSLMode.REQUIRED),
Arguments.of(
MySqlConnectorConfig.SecureConnectionMode.VERIFY_CA, SSLMode.VERIFY_CA),
Arguments.of(
MySqlConnectorConfig.SecureConnectionMode.VERIFY_IDENTITY,
SSLMode.VERIFY_IDENTITY));
}

private void assertJdbcUrl(String expected, String actual) {
// Compare after splitting to avoid the orderless jdbc parameters in jdbc url at Java 11
String[] expectedParam = expected.split("&");
Expand Down
Loading