Reactor Kafka 中的提交偏移量答案

【问题标题】：Commit offset in Reactor KafkaReactor Kafka 中的提交偏移量
【发布时间】：2021-11-14 11:26:49
【问题描述】：

我有一个反应堆 Kafka 项目，它使用来自 Kafka 主题的消息，转换消息然后写入另一个主题。

public Flux<String> consume(String destTopic) {
        return kafkaConsumerTemplate
                .receiveAutoAck()
                .doOnNext(consumerRecord -> log.info("received key={}, value={} from topic={}, offset={}",
                        consumerRecord.key(),
                        consumerRecord.value(),
                        consumerRecord.topic(),
                        consumerRecord.offset())
                )
                .doOnNext(s-> sendToKafka(s,destTopic))
                .map(ConsumerRecord::value)
                .doOnError(throwable -> log.error("Error while consuming : {}", throwable.getMessage()));
    }

我的理解是只有在反应器中成功完成所有序列步骤后才会提交偏移量。那是对的吗？我想确保不处理下一条记录，除非当前记录成功发送到目标 Kafka Topic。

【问题讨论】：

标签： apache-kafka spring-kafka project-reactor reactor-kafka

【解决方案1】：

实现是这样的：

@Override
public Flux<Flux<ConsumerRecord<K, V>>> receiveAutoAck(Integer prefetch) {
    return withHandler(AckMode.AUTO_ACK, (scheduler, handler) -> handler
        .receive()
        .filter(it -> !it.isEmpty())
        .publishOn(scheduler, preparePublishOnQueueSize(prefetch))
        .map(consumerRecords -> Flux.fromIterable(consumerRecords)
            .doAfterTerminate(() -> {
                for (ConsumerRecord<K, V> r : consumerRecords) {
                    handler.acknowledge(r);
                }
            })));
}

因此，每个ConsumerRecords 仅在其Flux 被完全处理时才被确认：成功或有错误。因此，它不是每条记录的提交。从技术上讲，无论如何它都不能是每条记录，因为我们只需要提交，我们的消费者应用程序就会失败，我们需要从之前留下的偏移量继续。当前活动的KafkaConsumer 在内存中保留一个游标，并不关心您是否提交。

如果您真的想要“每条记录”，请参阅 ReactiveKafkaConsumerTemplate.receive() 及其 KafkaReceiver.receive() 代表：

/**
 * Starts a Kafka consumer that consumes records from the subscriptions or partition
 * assignments configured for this receiver. Records are consumed from Kafka and delivered
 * on the returned Flux when requests are made on the Flux. The Kafka consumer is closed
 * when the returned Flux terminates.
 * <p>
 * Every record must be acknowledged using {@link ReceiverOffset#acknowledge()} in order
 * to commit the offset corresponding to the record. Acknowledged records are committed
 * based on the configured commit interval and commit batch size in {@link ReceiverOptions}.
 * Records may also be committed manually using {@link ReceiverOffset#commit()}.
 *
 * @return Flux of inbound receiver records that are committed only after acknowledgement
 */
default Flux<ReceiverRecord<K, V>> receive() {

【讨论】：

感谢@Artem Bilan 的解释。因此，如果我理解正确，在我的代码中，receiveAutoAck 将接收包含一批消费者记录的通量，消费者记录将在其整个通量被处理时提交，即在处理 sendtoKafka 之后以及它是否成功。对吗？
正确。独立地，如果 sendtoKafka() 成功与否。对于像接收-n-发送这样的管道用例，请考虑使用receiveExactlyOnce()。

【解决方案2】：

如果你想控制提交行为，你需要像这样禁用自动提交：

ReceiverOptions.create()
    .commitInterval(Duration.ZERO)             
    .commitBatchSize(0)

然后您需要在记录处理完毕后提交：

final ReceiverOptions<String, String> receiverOptions = ReceiverOptions.<String, String>create()
                .commitInterval(Duration.ZERO)
                .commitBatchSize(0)
                .subscription(List.of("mytopic"));
sender.send(KafkaReceiver.create(receiverOptions)
                        .receive()
                        .map(m -> SenderRecord.create(transform(m.key(), m.value()), m.receiverOffset()))) // transform the data
                .doOnNext(m -> m.correlationMetadata().commit().block()); // Synchronous commit after record is successfully delivered

【讨论】：