public static MySqlConnection createMySqlConnection(MySqlSourceConfig sourceConfig) {
        return createMySqlConnection(
                sourceConfig.getDbzConfiguration(), sourceConfig.getJdbcProperties());
    }
@Test
    void testCreateMySqlConnection() {
        // test without set useSSL
        Properties jdbcProps = new Properties();
        jdbcProps.setProperty("onlyTest", "test");
        MySqlSourceConfig configWithoutUseSSL = getConfig(jdbcProps);
        MySqlConnection connection0 = DebeziumUtils.createMySqlConnection(configWithoutUseSSL);
        assertJdbcUrl(
                "jdbc:mysql://localhost:3306/?useSSL=false&connectTimeout=20000&useInformationSchema=true"
                        + "&nullCatalogMeansCurrent=false&characterSetResults=UTF-8&onlyTest=test"
                        + "&zeroDateTimeBehavior=CONVERT_TO_NULL&characterEncoding=UTF-8&useUnicode=true",
                connection0.connectionString());

        // test with set useSSL=false
        jdbcProps.setProperty("useSSL", "false");
        MySqlSourceConfig configNotUseSSL = getConfig(jdbcProps);
        MySqlConnection connection1 = DebeziumUtils.createMySqlConnection(configNotUseSSL);
        assertJdbcUrl(
                "jdbc:mysql://localhost:3306/?connectTimeout=20000&useInformationSchema=true"
                        + "&nullCatalogMeansCurrent=false&characterSetResults=UTF-8&useSSL=false&onlyTest=test"
                        + "&zeroDateTimeBehavior=CONVERT_TO_NULL&characterEncoding=UTF-8&useUnicode=true",
                connection1.connectionString());

        // test with set useSSL=true
        jdbcProps.setProperty("useSSL", "true");
        MySqlSourceConfig configUseSSL = getConfig(jdbcProps);
        MySqlConnection connection2 = DebeziumUtils.createMySqlConnection(configUseSSL);
        assertJdbcUrl(
                "jdbc:mysql://localhost:3306/?connectTimeout=20000&useInformationSchema=true"
                        + "&nullCatalogMeansCurrent=false&characterSetResults=UTF-8&useSSL=true&onlyTest=test"
                        + "&zeroDateTimeBehavior=CONVERT_TO_NULL&characterEncoding=UTF-8&useUnicode=true",
                connection2.connectionString());
    }