KafkaConsumer#resetToLastCommittedPositions doesn't work with the default throttled ack strategy 🤷🏻‍♂️

[Dougniel]

Hi 🙋,

? Context

We process records in batch and we want to send only poisoned records to a dead-letter-queue in case of failure, not all the records in the batch 🙅.

For that we developed a custom failure strategy🫰to do a retry record per record ¹ in which we apply the dead-letter-queue's base implementation strategy. To do the retry, we choose to use KafkaConsumer#resetToLastCommittedPositions (accessed through KafkaClientService) that fits perfectly our needs 👌

But.. sometimes retrying a batch of records also replayed the previous one too, that is where we noticed that the throttled acknowledgement strategy was not aware of that resetToLastCommittedPositions: the pending acks are not flushed to Kafka.

In our case we switched to the latest acknowledgement strategy and it's ok for now🚶, but it looks incoherent behaviour IMO and could lead to some very bad experiences 😞

🙏🏼 What I would like

Regarding that :

• KafkaConsumer#resetToLastCommittedPositions is public and accessible through KafkaClientService)
• the throttled acknowledgement strategy is the default one and pretty useful (thanks guys 👍)

Would it will be possible to adapt the implementation of KafkaConsumer#resetToLastCommittedPositions to update the throttled mechanism accordingly ?

Seems that the actual partitions revocation mechanism could fit well :

1. flushing the pending acks
2. clearing off tracks of the received records not yet acknowledged

Thx

Footnotes

1. Switching a batch into records is done with a PublisherDecorator that transformToIterable the Multi ↩

[ozangunalp]

Thanks for the detailed issue report!

It does seem like an interesting use case.

Have you tried the smallrye-fault-tolerance @Retry on a method other than the @Incoming one?

Whether you are using the Kafka batching or your own batching policy, you can filter out poisoned records and nack them right away and you might be able to refactor your processing logic to another method that is annotated with @Retry.

Otherwise, if you really must implement a custom failure strategy, what you need is to be able to call the commit strategy from the failure one. By the way, you already have the reference to KafkaConsumer through the KafkaFailureStrategy.Factory create method, you don't need to fetch it through KafkaClientService.

[Dougniel]

Have you tried the smallrye-fault-tolerance @Retry on a method other than the @Incoming one?

Honestly speaking, I didn't invest so much time in this option as I felt lost at first even if it was my first thought, unfortunately I didn't find a way to split a batch message with such mechanism before the retry.

Whether you are using the Kafka batching or your own batching policy, you can filter out poisoned records and nack them right away and you might be able to refactor your processing logic to another method that is annotated with @Retry.

I didn't find any documentation about extension facilities in smallrye-fault-tolerance/@Retry as in smallrye-reactive-messaging, do you have some ideas ?

Otherwise, if you really must implement a custom failure strategy, what you need is to be able to call the commit strategy from the failure one. By the way, you already have the reference to KafkaConsumer through the KafkaFailureStrategy.Factory create method, you don't need to fetch it through KafkaClientService.

Right 👍