Implementing max retries with batching enabled

[CharlyRien]

Hello there,

First of all, thank you for your work,

I'm currently testing your project to see if it can fit in our current production environment.

For this, I would need to know how to properly implement a ParallelStreamProcessor (ordering by KEY) that handles batching and also has a limited number of retries for a given batch that failed. (I would also need to implement manually a kind of exponential retry.)

I read the documentation and it seems to say that once a batch has failed, it is retried but it is not guaranteed to have the same records in this batch.
So I wanted to know how I can be sure of the number of failed attempts on a given batch?

Thanks :-)

[rkolesnev]

Hi @CharlyRien,

I believe you are right in that batches can be reshuffled as they are taken out of processing into retry queue on error and work is redistributed for processing again once retry backoff is elapsed.
You can track number of previous retries per message in the user function and filter them out of the batch before submitting it for processing - for example:

        parallelConsumer.poll(recordContexts -> {
            //Filter out messages that already retried too many times
            List<PollContext<String, String>> recordsToProcessFiltered = recordContexts.stream()
                    .filter(recordContext -> recordContext.getNumberOfFailedAttempts() < maxRetries).collect(Collectors.toList());
            if(recordsToProcessFiltered.isEmpty()){ 
                //Nothing to process - just return
                //all messages will be dropped as processed.
                return; 
            }
            doProcessing(recordsToProcessFiltered);
        });

Only problem with the above approach - the whole batch will still fail until all the 'bad' messages are filtered out - so some of the messages can still keep being retried.

Other consideration, I guess - if you know that the message exceeded retries (in the delay function calculation) - you may want it to have next delay set to 0 - and let it get filtered out with the approach above (or send it to DLQ as well etc).

For retry delay / backoff - you can supply custom function through ParallelConsumerOptions.retryDelayProvider - and do variable delay based on number of retries for the message.

[CharlyRien]

Thanks, I'll try your approach.