MySqlSourceEnumerator类核心方法

start方法

 @Override
    public void start() {
        // 如果启动参数是INITIAL
       // splitAssigner 是MySqlHybridSplitAssigner, 否则是MySqlBinlogSplitAssigner
        // 启动 SplitAssigner
        splitAssigner.open();
        // 如果是新增同步表, 重新启动CDC任务场景下: 
       //当新添加的表的快照完成分配时，请求更新binlog的分割状态
        requestBinlogSplitUpdateIfNeed();
        // 周期性的执行syncWithReaders 方法
        this.context.callAsync(
                this::getRegisteredReader,
                this::syncWithReaders,
                CHECK_EVENT_INTERVAL,
                CHECK_EVENT_INTERVAL);
    }

 /**
     * 1、检查是否新添加的表的快照已经完成分配。
     * 2、如果是，则遍历已注册的读取器。
     * 3、对每个读取器，发送更新binlog分割状态的请求事件。
     */
    private void requestBinlogSplitUpdateIfNeed() {
        if (isNewlyAddedAssigningSnapshotFinished(splitAssigner.getAssignerStatus())) {
            for (int subtaskId : getRegisteredReader()) {
                LOG.info(
                        "The enumerator requests subtask {} to update the binlog split after newly added table.",
                        subtaskId);
                context.sendEventToSourceReader(subtaskId, new BinlogSplitUpdateRequestEvent());
            }
        }
    }

   /**
     * 当SourceEnumerator恢复或者SourceEnumerator和SourceReader之间的通信失败时，
     * 可能会错过一些通知事件。告诉所有的SourceReader报告它们已完成但未确认的分片。
     */
    private void syncWithReaders(int[] subtaskIds, Throwable t) {
        if (t != null) {
            throw new FlinkRuntimeException("Failed to list obtain registered readers due to:", t);
        }
        // when the SourceEnumerator restores or the communication failed between
        // SourceEnumerator and SourceReader, it may missed some notification event.
        // tell all SourceReader(s) to report there finished but unacked splits.
        if (splitAssigner.waitingForFinishedSplits()) {
            for (int subtaskId : subtaskIds) {
                context.sendEventToSourceReader(
                        subtaskId, new FinishedSnapshotSplitsRequestEvent());
            }
        }

        requestBinlogSplitUpdateIfNeed();
    }

其中splitAssigner.open()方法后续再分析

handleSplitRequest 方法

接收来自 reader 的请求，对数据进行分片,通过 assignSplits 对分片进行分配, 注意assignSplits方法在CK完成回调notifyCheckpointComplete方法中也会调用, 如果已经是split分配完成后的一个ck，就可以标志着全量阶段已经完成，可以开始准备下发BinlogSplit了

 @Override
    public void handleSplitRequest(int subtaskId, @Nullable String requesterHostname) {
        if (!context.registeredReaders().containsKey(subtaskId)) {
            // reader failed between sending the request and now. skip this request.
            return;
        }

        // 接收来自reader的分片请求, 根据reader的taskId, 放入到readersAwaitingSplit TreeSet中, 表示等待分配split的reader请求
        readersAwaitingSplit.add(subtaskId);
        // 为reader分配split, 分配完成后, 从readersAwaitingSplit集合中 remove掉
        assignSplits();
    }

     private void assignSplits() {
        final Iterator<Integer> awaitingReader = readersAwaitingSplit.iterator();

        while (awaitingReader.hasNext()) {
            int nextAwaiting = awaitingReader.next();
            // if the reader that requested another split has failed in the meantime, remove
            // it from the list of waiting readers
            // 判断当前注册的reader集合是否存在
            if (!context.registeredReaders().containsKey(nextAwaiting)) {
                awaitingReader.remove();
                continue;
            }

            // 判断当前的Snapshot全量阶段是否完成, 如果完成根据设置的closeIdleReaders参数, 是否关闭空闲Reader
            /**
             *  Flink CDC 的增量快照框架有两个主要阶段: 全量阶段和增量阶段。
             *  这两个阶段的并行度并不相同，全量阶段支持多并行度，加快大量数据的同步过程，
             *  增量阶段读取Binlog变更日志，需要使用单并发保证事件的顺序和正确性。
             *  在全量阶段读取结束后，由于增量阶段只需要一个并发，会出现大量的空闲 Reader，比较浪费资源。
             *  2.4 版本使用增量快照连接器时，支持配置打开自动关闭空闲 Reader 的功能来关闭这些空闲 Reader
             */
            if (splitAssigner.isStreamSplitAssigned()
                    && sourceConfig.isCloseIdleReaders()
                    && noMoreSnapshotSplits()
                    && (binlogSplitTaskId != null && !binlogSplitTaskId.equals(nextAwaiting))) {
                // close idle readers when snapshot phase finished.
                // 向reader 发送关闭请求, 释放资源
                context.signalNoMoreSplits(nextAwaiting);
                awaitingReader.remove();
                LOG.info("Close idle reader of subtask {}", nextAwaiting);
                continue;
            }

            // TODO: 待分析
            Optional<MySqlSplit> split = splitAssigner.getNext();
            if (split.isPresent()) {
                final MySqlSplit mySqlSplit = split.get();
                /**
                 *  向reader 分配MySqlSplit
                 */
                context.assignSplit(mySqlSplit, nextAwaiting);
                if (mySqlSplit instanceof MySqlBinlogSplit) {
                    this.binlogSplitTaskId = nextAwaiting;
                }
                awaitingReader.remove();
                LOG.info("The enumerator assigns split {} to subtask {}", mySqlSplit, nextAwaiting);
            } else {
                // there is no available splits by now, skip assigning
                requestBinlogSplitUpdateIfNeed();
                break;
            }
        }
    }

handleSourceEvent方法

处理 SourceEvent，是 SplitEnumerator 和 SourceReader 之间来回传递的自定义事件。可以利用此机制来执行复杂的协调任务
处理事件类型:

• FinishedSnapshotSplitsReportEvent: reader完成读取chunk数据事件
• BinlogSplitMetaRequestEvent: reader请求分配BinlogSplit
• BinlogSplitUpdateAckEvent: BinlogSplit分配确认ACK事件
• LatestFinishedSplitsNumberRequestEvent: 请求获取当前最新已完成的Snapshot chunk数量
• BinlogSplitAssignedEvent: reader已分配到BinlogSplit事件

 @Override
    public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
        // 接收到reader FinishedSnapshotSplitsReportEvent 事件
        if (sourceEvent instanceof FinishedSnapshotSplitsReportEvent) {
            LOG.info(
                    "The enumerator under {} receives finished split offsets {} from subtask {}.",
                    splitAssigner.getAssignerStatus(),
                    sourceEvent,
                    subtaskId);
            FinishedSnapshotSplitsReportEvent reportEvent =
                    (FinishedSnapshotSplitsReportEvent) sourceEvent;
            Map<String, BinlogOffset> finishedOffsets = reportEvent.getFinishedOffsets();

            // 在Snapshot全量阶段, 保存每个reader读取chunk完成时候的binlog点位
            splitAssigner.onFinishedSplits(finishedOffsets);
            requestBinlogSplitUpdateIfNeed();

            // send acknowledge event
            // 发送ack event
            FinishedSnapshotSplitsAckEvent ackEvent =
                    new FinishedSnapshotSplitsAckEvent(new ArrayList<>(finishedOffsets.keySet()));
            context.sendEventToSourceReader(subtaskId, ackEvent);
        } else if (sourceEvent instanceof BinlogSplitMetaRequestEvent) { //  处理BinlogSplitMetaRequestEvent事件请求
            LOG.debug(
                    "The enumerator receives request for binlog split meta from subtask {}.",
                    subtaskId);
            sendBinlogMeta(subtaskId, (BinlogSplitMetaRequestEvent) sourceEvent);
        } else if (sourceEvent instanceof BinlogSplitUpdateAckEvent) {
            LOG.info(
                    "The enumerator receives event that the binlog split has been updated from subtask {}. ",
                    subtaskId);
            splitAssigner.onBinlogSplitUpdated();
        } else if (sourceEvent instanceof LatestFinishedSplitsNumberRequestEvent) {
            LOG.info(
                    "The enumerator receives request from subtask {} for the latest finished splits number after added newly tables. ",
                    subtaskId);
            // 告诉reader 当前已完成的Snapshot FinishedSplitInfos size
            handleLatestFinishedSplitNumberRequest(subtaskId);
        } else if (sourceEvent instanceof BinlogSplitAssignedEvent) { // 处理BinlogSplitAssignedEvent事件
            LOG.info(
                    "The enumerator receives notice from subtask {} for the binlog split assignment. ",
                    subtaskId);
            binlogSplitTaskId = subtaskId;
        }
    }

至此已梳理MySqlSourceEnumerator类中的核心方法及作用, 但是全量阶段的chunk划分及向reader分配MySqlSplit逻辑在哪呢?

• chunk划分: 在start方法中的splitAssigner.open() 实现
• MySqlSplit分配: handleSplitRequest()->assignSplits()->splitAssigner.getNext()