[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/org/apache/flink/cdc/connectors/mysql/debezium/DebeziumUtils.java

@@ -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;
@@ -242,12 +243,30 @@ private static Map<String, String> querySystemVariables(
         return variables;
     }
 
+    static SSLMode sslModeFor(MySqlConnectorConfig.SecureConnectionMode mode) {
+        switch (mode) {
+            case DISABLED:
+                return SSLMode.DISABLED;
+            case PREFERRED:
+                return SSLMode.PREFERRED;
+            case REQUIRED:
+                return SSLMode.REQUIRED;
+            case VERIFY_CA:
+                return SSLMode.VERIFY_CA;
+            case VERIFY_IDENTITY:
+                return SSLMode.VERIFY_IDENTITY;
+        }
+        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());
         }

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