private void checkBinlogFormat(JdbcConnection connection) throws SQLException {
        String mode =
                connection
                        .queryAndMap(
                                "SHOW GLOBAL VARIABLES LIKE 'binlog_format'",
                                rs -> rs.next() ? rs.getString(2) : "")
                        .toUpperCase();
        if (!BINLOG_FORMAT_ROW.equals(mode)) {
            throw new ValidationException(
                    String.format(
                            "The MySQL server is configured with binlog_format %s rather than %s, which is "
                                    + "required for this connector to work properly. "
                                    + "Change the MySQL configuration to use a "
                                    + "binlog_format=ROW and restart the connector.",
                            mode, BINLOG_FORMAT_ROW));
        }
    }
Check whether the binlog format is ROW.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/MySqlValidator.java