private void maybeLogSplitChangesHandlingResult(SplitsChange<KafkaPartitionSplit> splitsChange) {
    if (LOG.isDebugEnabled()) {
      StringJoiner splitsInfo = new StringJoiner(",");
      for (KafkaPartitionSplit split : splitsChange.splits()) {
        long startingOffset =
            retryOnWakeup(
                () -> consumer.position(split.getTopicPartition()), "logging starting position");
        long stoppingOffset = getStoppingOffset(split.getTopicPartition());
        splitsInfo.add(
            String.format(
                "[%s, start:%d, stop: %d]",
                split.getTopicPartition(), startingOffset, stoppingOffset));
      }
      LOG.debug("SplitsChange handling result: {}", splitsInfo);
    }
  }
A {@link SplitReader} implementation that reads records from Kafka partitions.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/internals/KafkaPartitionSplitReader.java