@Override
    public void validate() {
        JdbcConnection connection = null;
        try {
            if (sourceConfig != null) {
                connection = DebeziumUtils.openJdbcConnection(sourceConfig);
            } else {
                // for the legacy source
                connection = DebeziumUtils.createMySqlConnection(Configuration.from(dbzProperties));
            }
            checkVersion(connection);
            checkBinlogFormat(connection);
            checkBinlogRowImage(connection);
        } catch (SQLException ex) {
            throw new TableException(
                    "Unexpected error while connecting to MySQL and validating", ex);
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    throw new FlinkRuntimeException("Closing connection error", e);
                }
            }
        }
        LOG.info("MySQL validation passed.");
    }
The validator for MySql: it only cares about the version of the database is larger than or equal
to 5.7. It also requires the binlog format in the database is ROW and row image is FULL.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/MySqlValidator.java