API потребителя реактора Кафки

Question

Мне нужно реализовать функцию повтора для потребителя кафки.Я использую пружинный реактор API.

Flux> inboundFlux = KafkaReceiver.create (receiveOptions) .receive ();

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

Параметр повтора для inboundFlux.retry () не работает.Подскажите пожалуйста как мне сделать эту работу.