Motivation:

The developer expects to get the KafkaHeaders.DELIVERY_ATTEMPT header from the received ConsumerRecord.
It works well with the SingleRecordListener.
However, it does not work well with the BatchListener even when containerProperties.setDeliveryAttemptHeader(true) is set.
If the KafkaContainerListener can include the KafkaHeaders.DELIVERY_ATTEMPT header for each ConsumerRecord with the BatchListener, it will be more helpful to people who they expect.

Modifications

• Add DeliveryAttemptAwareRetryListener class.
• Add unit tests and integration tests for DeliveryAttemptAwareRetryListener.

To Reviewer

With current spring-kafka codes, it is hard to integrate it to FailedRecordTracker.
By the way, the function ErrorHandlingUtils.retryBatch() has deliveryAttempt count as their local scope and has same reference for ConsumerRecord.
Thus, the DeliveryAttemptAwareRetryListener is easier way to achieve the goal.
Because, in each fail to invocation, ErrorHandlingUtils.retryBatch() calls listeners.forEach(listener -> listener.failedDelivery(records, thrownException, attempt).

If the developer want to use this feature, they should create ContainerFactory and DeliveryAttemptAwareRetryListener. also, they set DeliveryAttemptAwareRetryListener as RetryListener.
Please refer to code below.

final FixedBackOff fixedBackOff = new FixedBackOff(1000L, MAX_ATTEMPT_COUNT1);
DefaultErrorHandler errorHandler = new DefaultErrorHandler(fixedBackOff);
errorHandler.setRetryListeners(new DeliveryAttemptAwareRetryListener());

ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory);
factory.setCommonErrorHandler(errorHandler);

Result:

• Fixes KafkaHeaders.DELIVERY_ATTEMPT is not added for batch listeners #3407