After retry no messages delivered to consumer about reactor-kafka

[OlegDokuka]

@FcoJavierSainz Can you please doublecheck that issue with the snapshot version (1.0.1.BUILD-SNAPSHOT)

[hfgbarrigas]

Any news on this bug? A bit critical since it stalls consumers and forces a restart.

[bsideup]

@rajinisivaram @smaldini we observe the same issue, do you have any update on it?
Seems to not comply with the Reactive spec

[bsideup]

After a discussion with @smaldini it seems that there is a workaround:

KafkaReceiver receiver =  KafkaReceiver.create(receiverOptions(Collections.singletonList(topic));
return Flux.defer(receiver::receive)
  .map(r -> { 
      if (atomicCounter.getAndIncrement() > 10)  {
          throw new RuntimeException("TEST");
      }
      r.receiverOffset().commit().block();
      return r;
  }
  .retry()
  .subscribe(r ->System.out.print(r.value()));

Thank for the work around...Any updates on fixing the retry?

[jiallombardo]

Also, the workaround appears to not work for the second retry. E.g. the receiver is restarted the first time a map function fails, and messages are consumed again; however, if the map function fails a second time, the messages aren't consumed anymore.

[FcoJavierSainz]

After one retry, stop again :(

[FcoJavierSainz]

@OlegDokuka It is working for me :) Thanks

[OlegDokuka]

@FcoJavierSainz, amazing! Thank you for doublechecking that!!! Close since fixed by #35

[jimhorng]

UPDATED
Flux.defer(receiver::receive) works for me, I found the root cause is that consumer's flux's scheduler not getting re-init on 2nd attempt, but the .defer() does that. FYR.

---

Seemed I'm facing the similar problem on using 1.0.1.RELEASE

• The 1st attempt it got below, which seems expected since I intentionally throw a RuntimeException

doFinally: signalType:cancel
doOnError:class java.lang.RuntimeException
doOnError:class reactor.core.Exceptions$ReactorRejectedExecutionException

• it does retry (show client config initiating in log), but in 2nd attempt and after, it got below error and does NOT receive message anymore :(

doFinally: signalType:onError
doOnError:class reactor.core.Exceptions$ReactorRejectedExecutionException

my code looks something like this:

KafkaReceiver<String, byte[]> kafkaReceiver = KafkaReceiver.create(options);
Flux<ReceiverRecord<String, byte[]>> kafkaFlux = kafkaReceiver.receive();
return kafkaFlux
        .doFinally(signalType -> {
            log.info("doFinally: signalType:{}", signalType.toString());
        })
        .map(record -> {
            ReceiverOffset offset = record.receiverOffset();
            processEvent(record, offset);
            offset.acknowledge();
            return record;
        })
        .doOnError(error -> {
            log.info("doOnError:{}", error.getClass());
            this.counter.set(0);
        })
        .retry(3)
        .subscribe()

Please advise, thanks :)

[ligasgr]

I'm also seeing the same issue on 1.0.1.
The defer() workaround does work but that is only a workaround.

[jli207]

We are also seeing this issue. Will this get fixed any time soon? Thanks.

[abhimanyuseth]

I had same issue and found this solution on StackOverflow worked for me
https://stackoverflow.com/questions/54984724/reactor-kafka-at-least-once-handling-failures-and-offsets-in-multi-partition

[billydh]

Will this be fixed in 1.3 RELEASE?

[bsideup]

Should be fixed in 1.3.0.RELEASE