GH-2542: FallbackBatchErrorHandler Improvement

Resolves #2542

Fix retry logic causing consumer leaving the group

**Short problem description**: consumer leaves the group when combined processing+backoff time is higher than `max.poll.interval.ms`.
**Root cause**: The retry logic in `ErrorHandlingUtils` does not call `poll()` (on the paused consumer) before re-trying the listener runnable, it calls `poll()` only before backing off. So if backoffInterval + duration of the following retried execution is longer than `max.poll.interval` - consumer leaves the group.
**Solution**: Amend `ErrorHandlingUtils` to have an extra call to `consumer.poll` right before retrying the listener runnable.

cleanup

fix import style

fix import style

cleanup

Author: truepele

spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandlingUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2022 the original author or authors.
+ * Copyright 2021-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -37,6 +37,7 @@
  * Utilities for error handling.
  *
  * @author Gary Russell
+ * @author Andrii Pelesh
  * @since 2.8
  *
  */
@@ -96,6 +97,7 @@ public static void retryBatch(Exception thrownException, ConsumerRecords<?, ?> r
 				if (!container.isRunning()) {
 					throw new KafkaException("Container stopped during retries");
 				}
+				consumer.poll(Duration.ZERO);
 				try {
 					invokeListener.run();
 					return;

spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -82,7 +82,7 @@ void recover() {
 		assertThat(this.invoked).isEqualTo(3);
 		assertThat(recovered).hasSize(2);
 		verify(consumer).pause(any());
-		verify(consumer, times(3)).poll(any());
+		verify(consumer, times(2 * this.invoked)).poll(any());
 		verify(consumer).resume(any());
 		verify(consumer, times(2)).assignment();
 		verifyNoMoreInteractions(consumer);
@@ -108,7 +108,7 @@ void successOnRetry() {
 		assertThat(this.invoked).isEqualTo(1);
 		assertThat(recovered).hasSize(0);
 		verify(consumer).pause(any());
-		verify(consumer).poll(any());
+		verify(consumer, times(2)).poll(any());
 		verify(consumer).resume(any());
 		verify(consumer, times(2)).assignment();
 		verifyNoMoreInteractions(consumer);
@@ -139,7 +139,7 @@ void recoveryFails() {
 		assertThat(this.invoked).isEqualTo(3);
 		assertThat(recovered).hasSize(1);
 		verify(consumer).pause(any());
-		verify(consumer, times(3)).poll(any());
+		verify(consumer, times(2 * this.invoked)).poll(any());
 		verify(consumer).resume(any());
 		verify(consumer, times(2)).assignment();
 		verify(consumer).seek(new TopicPartition("foo", 0), 0L);
@@ -208,9 +208,11 @@ void rePauseOnRebalance() {
 		inOrder.verify(container).publishConsumerPausedEvent(map.keySet(), "For batch retry");
 		inOrder.verify(consumer).poll(any());
 		inOrder.verify(consumer).pause(any());
+		inOrder.verify(consumer).poll(any());
+		inOrder.verify(consumer).pause(any());
 		inOrder.verify(consumer).resume(any());
 		inOrder.verify(container).publishConsumerResumedEvent(map.keySet());
-		verify(consumer, times(3)).assignment();
+		verify(consumer, times(4)).assignment();
 		verifyNoMoreInteractions(consumer);
 		assertThat(pubPauseCalled.get()).isTrue();
 	}