@Override
  protected void onSplitFinished(Map<String, KafkaPartitionSplitState> finishedSplitIds) {
    finishedSplitIds.forEach(
        (ignored, splitState) -> {
          if (splitState.getCurrentOffset() >= 0) {
            offsetsOfFinishedSplits.put(
                splitState.getTopicPartition(),
                new OffsetAndMetadata(splitState.getCurrentOffset()));
          }
        });
  }
The source reader for Kafka partitions.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/internals/KafkaSourceReader.java