private void maybeRegisterKafkaConsumerMetrics(
      Properties props,
      KafkaSourceReaderMetrics kafkaSourceReaderMetrics,
      KafkaConsumer<?, ?> consumer) {
    final Boolean needToRegister =
        KafkaSourceOptions.getOption(
            props, KafkaSourceOptions.REGISTER_KAFKA_CONSUMER_METRICS, Boolean::parseBoolean);
    if (needToRegister) {
      kafkaSourceReaderMetrics.registerKafkaConsumerMetrics(consumer);
    }
  }
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