private void handleBatch(List<ChangeEvent<SourceRecord, SourceRecord>> changeEvents, boolean isStreamingPhase)
            throws Exception {
        if (CollectionUtils.isEmpty(changeEvents)) {
            return;
        }
        this.processTime = System.currentTimeMillis();

        for (ChangeEvent<SourceRecord, SourceRecord> event : changeEvents) {
            SourceRecord record = event.value();
            updateMessageTimestamp(record);
            fetchDelay = isInDbSnapshotPhase ? 0L : processTime - messageTimestamp;

            if (isHeartbeatEvent(record)) {
                // keep offset update
                synchronized (checkpointLock) {
                    debeziumOffset.setSourcePartition(record.sourcePartition());
                    debeziumOffset.setSourceOffset(record.sourceOffset());
                }
            }
            deserialization.deserialize(record, debeziumCollector, isStreamingPhase);

            if (isHeartbeatEvent(record)) {
                // drop heartbeat events
                continue;
            }
            if (!isSnapshotRecord(record)) {
                LOG.debug("Snapshot phase finishes.");
                isInDbSnapshotPhase = false;
            }
            // emit the actual records. this also updates offset state atomically
            emitRecordsUnderCheckpointLock(
                    debeziumCollector.records, record.sourcePartition(), record.sourceOffset());
        }
    }
The metric indicates delay from data generation to leaving the source operator.

<p>Note: the metric is available during the binlog phase. Use 0 to indicate the metric is
unavailable.</p>
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mongodb-cdc/src/main/java/org/apache/inlong/sort/cdc/mongodb/debezium/internal/DebeziumChangeFetcher.java