在产生事件 B 后手动确认 Kafka 事件 A 消费

Question

我有一种情况，我必须消耗事件 A 并进行一些处理，然后生成事件 B。所以我的问题是处理崩溃并且应用程序在消耗已经消耗 A 时无法生成 B。我的方法是在成功发布 B 后确认，我是正确的还是应该针对这种情况实施其他解决方案？

@KafkaListener(
        id = TOPIC_ID,
        topics = TOPIC_ID,
        groupId = GROUP_ID,
        containerFactory = LISTENER_CONTAINER_FACTORY
)
public void listen(List<Message<A>> messages, Acknowledgment acknowledgment) {

    try {
        final AEvent aEvent = messages.stream()
                .filter(message -> null != message.getPayload())
                .map(Message::getPayload)
                .findFirst()
                .get();

        processDao.doSomeProcessing() // returns a Mono<Example> by calling an externe API
                .subscribe(
                        response -> {
                            ProducerRecord<String, BEvent> BEventRecord = new ProducerRecord<>(TOPIC_ID, null, BEvent);

                            ListenableFuture<SendResult<String, BEvent>> future = kafkaProducerTemplate.send(buildBEvent());
                            future.addCallback(new ListenableFutureCallback<SendResult<String, BEvent>>() {
                                @Override
                                public void onSuccess(SendResult<String, BEvent> BEventSendResult) {
                                    //TODO: do when event published successfully
                                }

                                @Override
                                public void onFailure(Throwable exception) {
                                    exception.printStackTrace();
                                    throw new ExampleException();
                                }
                            });
                        },
                        error -> {
                            error.printStackTrace();
                            throw new ExampleException();
                        }
                );
        acknowledgment.acknowledge(); // ??
    } catch (ExampleException) {
        exception.printStackTrace();
    }
}

Answer 1

在使用 reactor 等异步代码时，您无法管理 kafka “确认”。

Kafka 不管理每个主题/分区的离散确认，只管理分区的最后提交偏移量。

如果您异步处理两条记录，您将争夺首先提交哪个偏移量。

您需要在侦听器容器线程上执行发送以保持正确的顺序。