@Override
    public void setProducerThrowable(Throwable producerThrowable) {
        if (isTableNotFoundException(producerThrowable)) {
            Matcher matcher =
                    NOT_FOUND_TABLE_MSG_PATTERN.matcher(producerThrowable.getCause().getMessage());
            String databaseName = matcher.group(1);
            String tableName = matcher.group(2);
            TableId tableId = new TableId(databaseName, null, tableName);
            if (context.getSchema().schemaFor(tableId) == null) {
                LOG.warn("Schema for table " + tableId + " is null");
                return;
            }
        }

        if (isSchemaOutOfSyncException(producerThrowable)) {
            super.setProducerThrowable(
                    new SchemaOutOfSyncException(
                            "Internal schema representation is probably out of sync with real database schema. "
                                    + "The reason could be that the table schema was changed after the starting "
                                    + "binlog offset, which is not supported when startup mode is set to "
                                    + sourceConfig.getStartupOptions().startupMode,
                            producerThrowable));
            return;
        }

        super.setProducerThrowable(producerThrowable);
    }
A subclass implementation of {@link ErrorHandler} which filter some {@link DebeziumException}, we
use this class instead of {@link io.debezium.connector.mysql.MySqlErrorHandler}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/debezium/task/context/MySqlErrorHandler.java