@Override
    public void dispatchSchemaChangeEvent(
            Collection<TableId> dataCollectionIds,
            SchemaChangeEventEmitter schemaChangeEventEmitter)
            throws InterruptedException {
        boolean anyNonfilteredEvent = false;
        if (dataCollectionIds == null || dataCollectionIds.isEmpty()) {
            anyNonfilteredEvent = true;
        } else {
            for (TableId dataCollectionId : dataCollectionIds) {
                if (filter.isIncluded(dataCollectionId)) {
                    anyNonfilteredEvent = true;
                    break;
                }
            }
        }
        if (!anyNonfilteredEvent) {
            // TODO `HistorizedSchema.storeOnlyCapturedTables` first appeared in io.debezium.core: 1.6.4-final,
            // which conflicts with io.debezium.core: 15.4-final of cdc-base,
            // so we implement this method separately in the sub-module.
            // if (historizedSchema == null || historizedSchema.storeOnlyCapturedTables()) {
            if (historizedSchema == null || historizedSchema.storeOnlyMonitoredTables()) {
                LOG.trace("Filtering schema change event for {}", dataCollectionIds);
                return;
            }
        }

        schemaChangeEventEmitter.emitSchemaChangeEvent(new SchemaChangeEventReceiver());
    }
A subclass implementation of {@link EventDispatcher}.

<pre>
 1. This class shares one {@link ChangeEventQueue} between multiple readers.
 2. This class override some methods for dispatching {@link HistoryRecord} directly,
    this is useful for downstream to deserialize the {@link HistoryRecord} back.
</pre>
Copy from com.ververica:flink-cdc-base:2.3.0.
inlong-sort/sort-flink/cdc-base/src/main/java/org/apache/inlong/sort/cdc/base/relational/JdbcSourceEventDispatcher.java