private void checkVersion(JdbcConnection connection) throws SQLException {
        String version =
                connection.queryAndMap("SELECT VERSION()", rs -> rs.next() ? rs.getString(1) : "");

        // Only care about the major version and minor version
        Integer[] versionNumbers =
                Arrays.stream(version.split("\\."))
                        .limit(2)
                        .map(Integer::new)
                        .toArray(Integer[]::new);
        boolean isSatisfied;
        if (versionNumbers[0] > 5) {
            isSatisfied = true;
        } else if (versionNumbers[0] < 5) {
            isSatisfied = false;
        } else {
            isSatisfied = versionNumbers[1] >= 6;
        }
        if (!isSatisfied) {
            throw new ValidationException(
                    String.format(
                            "Currently Flink MySql CDC connector only supports MySql "
                                    + "whose version is larger or equal to 5.6, but actual is %s.%s.",
                            versionNumbers[0], versionNumbers[1]));
        }
    }
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