private void syncWithReaders(int[] subtaskIds, Throwable t) {
        if (t != null) {
            throw new FlinkRuntimeException("Failed to list obtain registered readers due to:", t);
        }
        // when the IncrementalSourceEnumerator restores or the communication failed between
        // IncrementalSourceEnumerator and JdbcIncrementalSourceReader, it may missed some
        // notification
        // event.
        // tell all JdbcIncrementalSourceReader(s) to report there finished but unacked splits.
        if (splitAssigner.waitingForFinishedSplits()) {
            for (int subtaskId : subtaskIds) {
                context.sendEventToSourceReader(
                        subtaskId, new FinishedSnapshotSplitsRequestEvent());
            }
        }
    }
Incremental source enumerator that enumerates receive the split request and assign the split to
source readers.
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/source/enumerator/IncrementalSourceEnumerator.java