project-reactor - Kafka 反应器消费者 API

Question

我需要在 kafka 消费者中实现重试功能。我正在使用弹簧反应器 api。

通量> inboundFlux = KafkaReceiver.create(receiverOptions).receive();

    inboundFlux.subscribe(r -> {
        if (accept(r)) {
            r.receiverOffset().acknowledge();
        }
    }, this::errorOut);

inboundFlux.retry() 上的重试选项不起作用。请建议我如何使这项工作。

标签： project-reactorspring-kafka