@Override
  public SourceReader<RowData, KafkaPartitionSplit> createReader(
      SourceReaderContext readerContext) {
    FutureCompletingBlockingQueue<RecordsWithSplitIds<ConsumerRecord<byte[], byte[]>>>
        elementsQueue = new FutureCompletingBlockingQueue<>();
    LogSourceHelper logReadHelper = logRetractionEnable ? new LogSourceHelper() : null;

    final KafkaSourceReaderMetrics kafkaSourceReaderMetrics =
        new KafkaSourceReaderMetrics(readerContext.metricGroup());
    Supplier<LogKafkaPartitionSplitReader> splitReaderSupplier =
        () ->
            new LogKafkaPartitionSplitReader(
                props,
                readerContext,
                kafkaSourceReaderMetrics,
                schema,
                logRetractionEnable,
                logReadHelper,
                logConsumerChangelogMode);
    LogKafkaRecordEmitter recordEmitter = new LogKafkaRecordEmitter(null);

    return new LogKafkaSourceReader<>(
        elementsQueue,
        new KafkaSourceFetcherManager(
            elementsQueue,
            splitReaderSupplier::get,
            (ignore) -> {},
            readerContext.getConfiguration()),
        recordEmitter,
        toConfiguration(props),
        readerContext,
        kafkaSourceReaderMetrics,
        logReadHelper);
  }
Get a logKafkaSourceBuilder to build a {@link LogKafkaSource}.

@return a Log Kafka source builder.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/read/source/log/kafka/LogKafkaSource.java