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

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlConnection.java

@@ -233,7 +233,7 @@ protected void setSystemProperty(String property, Field field, boolean showValue
      *
      * @return the session variables that are related to sessions ssl version
      */
-    protected String getSessionVariableForSslVersion() {
+    public String getSessionVariableForSslVersion() {
         final String sslVersion = "Ssl_version";
         LOGGER.debug("Reading MySQL Session variable for Ssl Version");
         Map<String, String> sessionVariables =

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java

@@ -27,6 +27,9 @@
 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.DefaultSSLSocketFactory;
+import com.github.shyiko.mysql.binlog.network.SSLMode;
+import com.github.shyiko.mysql.binlog.network.SSLSocketFactory;
 import io.debezium.config.Configuration;
 import io.debezium.connector.mysql.MySqlConnection;
 import io.debezium.connector.mysql.MySqlConnectorConfig;
@@ -46,7 +49,20 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+
 import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.X509Certificate;
 import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -56,6 +72,8 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.function.Predicate;
 
+import static io.debezium.util.Strings.isNullOrEmpty;
+
 /** Utilities related to Debezium. */
 public class DebeziumUtils {
     private static final String QUOTED_CHARACTER = "`";
@@ -93,13 +111,27 @@ public static MySqlConnection createMySqlConnection(
     }
 
     /** Creates a new {@link BinaryLogClient} for consuming mysql binlog. */
-    public static BinaryLogClient createBinaryClient(Configuration dbzConfiguration) {
+    public static BinaryLogClient createBinaryClient(
+            Configuration dbzConfiguration, MySqlConnection connection) {
         final MySqlConnectorConfig connectorConfig = new MySqlConnectorConfig(dbzConfiguration);
-        return new BinaryLogClient(
-                connectorConfig.hostname(),
-                connectorConfig.port(),
-                connectorConfig.username(),
-                connectorConfig.password());
+        BinaryLogClient client =
+                new BinaryLogClient(
+                        connectorConfig.hostname(),
+                        connectorConfig.port(),
+                        connectorConfig.username(),
+                        connectorConfig.password());
+        SSLMode sslMode = sslModeFor(connectorConfig.sslMode());
+        if (sslMode != null) {
+            client.setSSLMode(sslMode);
+        }
+        if (connectorConfig.sslModeEnabled()) {
+            SSLSocketFactory sslSocketFactory =
+                    getBinlogSslSocketFactory(connectorConfig, connection);
+            if (sslSocketFactory != null) {
+                client.setSslSocketFactory(sslSocketFactory);
+            }
+        }
+        return client;
     }
 
     /** Creates a new {@link MySqlDatabaseSchema} to monitor the latest MySql database schemas. */
@@ -242,12 +274,101 @@ 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) {
+            throw new FlinkRuntimeException(
+                    String.format("Invalid SecureConnectionMode provided: %s ", mode.name()), e);
+        }
+    }
+
+    // see
+    // flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource#getBinlogSslSocketFactory
+    static SSLSocketFactory getBinlogSslSocketFactory(
+            MySqlConnectorConfig connectorConfig, MySqlConnection connection) {
+        String acceptedTlsVersion = connection.getSessionVariableForSslVersion();
+        if (!isNullOrEmpty(acceptedTlsVersion)) {
+            SSLMode sslMode = sslModeFor(connectorConfig.sslMode());
+            LOG.info(
+                    "Enable ssl {} mode for connector {}",
+                    sslMode,
+                    connectorConfig.getLogicalName());
+
+            final char[] keyPasswordArray = connection.connectionConfig().sslKeyStorePassword();
+            final String keyFilename = connection.connectionConfig().sslKeyStore();
+            final char[] trustPasswordArray = connection.connectionConfig().sslTrustStorePassword();
+            final String trustFilename = connection.connectionConfig().sslTrustStore();
+            KeyManager[] keyManagers = null;
+            if (keyFilename != null) {
+                try {
+                    KeyStore ks = connection.loadKeyStore(keyFilename, keyPasswordArray);
+
+                    KeyManagerFactory kmf = KeyManagerFactory.getInstance("NewSunX509");
+                    kmf.init(ks, keyPasswordArray);
+
+                    keyManagers = kmf.getKeyManagers();
+                } catch (KeyStoreException
+                        | NoSuchAlgorithmException
+                        | UnrecoverableKeyException e) {
+                    throw new FlinkRuntimeException("Could not load keystore", e);
+                }
+            }
+            TrustManager[] trustManagers;
+            try {
+                KeyStore ks = null;
+                if (trustFilename != null) {
+                    ks = connection.loadKeyStore(trustFilename, trustPasswordArray);
+                }
+
+                if (ks == null && (sslMode == SSLMode.PREFERRED || sslMode == SSLMode.REQUIRED)) {
+                    trustManagers =
+                            new TrustManager[] {
+                                new X509TrustManager() {
+
+                                    @Override
+                                    public void checkClientTrusted(
+                                            X509Certificate[] x509Certificates, String s) {}
+
+                                    @Override
+                                    public void checkServerTrusted(
+                                            X509Certificate[] x509Certificates, String s) {}
+
+                                    @Override
+                                    public X509Certificate[] getAcceptedIssuers() {
+                                        return new X509Certificate[0];
+                                    }
+                                }
+                            };
+                } else {
+                    TrustManagerFactory tmf =
+                            TrustManagerFactory.getInstance(
+                                    TrustManagerFactory.getDefaultAlgorithm());
+                    tmf.init(ks);
+                    trustManagers = tmf.getTrustManagers();
+                }
+            } catch (KeyStoreException | NoSuchAlgorithmException e) {
+                throw new FlinkRuntimeException("Could not load truststore", e);
+            }
+            // DBZ-1208 Resembles the logic from the upstream BinaryLogClient, only that
+            // the accepted TLS version is passed to the constructed factory
+            final KeyManager[] finalKMS = keyManagers;
+            return new DefaultSSLSocketFactory(acceptedTlsVersion) {
+
+                @Override
+                protected void initSSLContext(SSLContext sc) throws GeneralSecurityException {
+                    sc.init(finalKMS, trustManagers, null);
+                }
+            };
+        }
+
+        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());
+                createBinaryClient(mySqlSourceConfig.getDbzConfiguration(), connection);
         if (mySqlSourceConfig.getServerIdRange() != null) {
             client.setServerId(mySqlSourceConfig.getServerIdRange().getStartServerId());
         }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java

@@ -39,6 +39,7 @@
 import com.github.shyiko.mysql.binlog.event.Event;
 import com.github.shyiko.mysql.binlog.event.EventType;
 import io.debezium.connector.base.ChangeEventQueue;
+import io.debezium.connector.mysql.MySqlConnection;
 import io.debezium.connector.mysql.MySqlStreamingChangeEventSourceMetrics;
 import io.debezium.pipeline.DataChangeEvent;
 import io.debezium.relational.TableId;
@@ -96,12 +97,15 @@ public class BinlogSplitReader implements DebeziumReader<SourceRecords, MySqlSpl
     private static final long READER_CLOSE_TIMEOUT = 30L;
 
     public BinlogSplitReader(MySqlSourceConfig sourceConfig, int subtaskId) {
-        this(
-                new StatefulTaskContext(
-                        sourceConfig,
-                        createBinaryClient(sourceConfig.getDbzConfiguration()),
-                        createMySqlConnection(sourceConfig)),
-                subtaskId);
+        this(createStatefulTaskContext(sourceConfig), subtaskId);
+    }
+
+    private static StatefulTaskContext createStatefulTaskContext(MySqlSourceConfig sourceConfig) {
+        MySqlConnection connection = createMySqlConnection(sourceConfig);
+        return new StatefulTaskContext(
+                sourceConfig,
+                createBinaryClient(sourceConfig.getDbzConfiguration(), connection),
+                connection);
     }
 
     public BinlogSplitReader(StatefulTaskContext statefulTaskContext, int subtaskId) {

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/SnapshotSplitReader.java

@@ -37,6 +37,7 @@
 
 import io.debezium.config.Configuration;
 import io.debezium.connector.base.ChangeEventQueue;
+import io.debezium.connector.mysql.MySqlConnection;
 import io.debezium.connector.mysql.MySqlConnectorConfig;
 import io.debezium.connector.mysql.MySqlOffsetContext;
 import io.debezium.connector.mysql.MySqlStreamingChangeEventSourceMetrics;
@@ -99,13 +100,15 @@ public class SnapshotSplitReader implements DebeziumReader<SourceRecords, MySqlS
 
     public SnapshotSplitReader(
             MySqlSourceConfig sourceConfig, int subtaskId, SnapshotPhaseHooks hooks) {
-        this(
-                new StatefulTaskContext(
-                        sourceConfig,
-                        createBinaryClient(sourceConfig.getDbzConfiguration()),
-                        createMySqlConnection(sourceConfig)),
-                subtaskId,
-                hooks);
+        this(createStatefulTaskContext(sourceConfig), subtaskId, hooks);
+    }
+
+    private static StatefulTaskContext createStatefulTaskContext(MySqlSourceConfig sourceConfig) {
+        MySqlConnection connection = createMySqlConnection(sourceConfig);
+        return new StatefulTaskContext(
+                sourceConfig,
+                createBinaryClient(sourceConfig.getDbzConfiguration(), connection),
+                connection);
     }
 
     public SnapshotSplitReader(

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtilsTest.java

@@ -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 {
@@ -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("&");