@Override
    public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
        if (sourceEvent instanceof FinishedSnapshotSplitsReportEvent) {
            LOG.info(
                    "The enumerator receives finished split offsets {} from subtask {}.",
                    sourceEvent,
                    subtaskId);
            FinishedSnapshotSplitsReportEvent reportEvent =
                    (FinishedSnapshotSplitsReportEvent) sourceEvent;
            Map<String, Offset> finishedOffsets = reportEvent.getFinishedOffsets();
            splitAssigner.onFinishedSplits(finishedOffsets);
            // send acknowledge event
            FinishedSnapshotSplitsAckEvent ackEvent =
                    new FinishedSnapshotSplitsAckEvent(new ArrayList<>(finishedOffsets.keySet()));
            context.sendEventToSourceReader(subtaskId, ackEvent);
        } else if (sourceEvent instanceof StreamSplitMetaRequestEvent) {
            LOG.debug(
                    "The enumerator receives request for stream split meta from subtask {}.",
                    subtaskId);
            sendStreamMetaRequestEvent(subtaskId, (StreamSplitMetaRequestEvent) sourceEvent);
        }
    }
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