private void informAboutUnknownTableIfRequired(Event event, TableId tableId, String typeToLog) {
        if (tableId != null && connectorConfig.getTableFilters().dataCollectionFilter().isIncluded(tableId)) {
            metrics.onErroneousEvent("source = " + tableId + ", event " + event);
            EventHeaderV4 eventHeader = event.getHeader();

            if (inconsistentSchemaHandlingMode == EventProcessingFailureHandlingMode.FAIL) {
                LOGGER.error(
                        "Encountered change event '{}' at offset {} for table {} whose schema isn't known to this connector. One possible cause is an incomplete database history topic. Take a new snapshot in this case.{}"
                                + "Use the mysqlbinlog tool to view the problematic event: mysqlbinlog --start-position={} --stop-position={} --verbose {}",
                        event, offsetContext.getOffset(), tableId, System.lineSeparator(), eventHeader.getPosition(),
                        eventHeader.getNextPosition(), offsetContext.getSource().binlogFilename());
                throw new DebeziumException("Encountered change event for table " + tableId
                        + " whose schema isn't known to this connector");
            } else if (inconsistentSchemaHandlingMode == EventProcessingFailureHandlingMode.WARN) {
                LOGGER.warn(
                        "Encountered change event '{}' at offset {} for table {} whose schema isn't known to this connector. One possible cause is an incomplete database history topic. Take a new snapshot in this case.{}"
                                + "The event will be ignored.{}"
                                + "Use the mysqlbinlog tool to view the problematic event: mysqlbinlog --start-position={} --stop-position={} --verbose {}",
                        event, offsetContext.getOffset(), tableId, System.lineSeparator(), System.lineSeparator(),
                        eventHeader.getPosition(), eventHeader.getNextPosition(),
                        offsetContext.getSource().binlogFilename());
            } else {
                LOGGER.debug(
                        "Encountered change event '{}' at offset {} for table {} whose schema isn't known to this connector. One possible cause is an incomplete database history topic. Take a new snapshot in this case.{}"
                                + "The event will be ignored.{}"
                                + "Use the mysqlbinlog tool to view the problematic event: mysqlbinlog --start-position={} --stop-position={} --verbose {}",
                        event, offsetContext.getOffset(), tableId, System.lineSeparator(), System.lineSeparator(),
                        eventHeader.getPosition(), eventHeader.getNextPosition(),
                        offsetContext.getSource().binlogFilename());
            }
        } else {
            LOGGER.debug("Filtering {} event: {} for non-monitored table {}", typeToLog, event, tableId);
            metrics.onFilteredEvent("source = " + tableId);
        }
    }
If we receive an event for a table that is monitored but whose metadata we
don't know, either ignore that event or raise a warning or error as per the
{@link MySqlConnectorConfig#INCONSISTENT_SCHEMA_HANDLING_MODE} configuration.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/io/debezium/connector/mysql/MySqlStreamingChangeEventSource.java