GH-3638: Fixes bug caused by race condition during handleAsyncFailure()

Fixes: #3638 

Issue: #3638

Author: chickenchickenlove

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

@@ -1466,7 +1466,14 @@ protected void pollAndInvoke() {
 
 		protected void handleAsyncFailure() {
 			List<FailedRecordTuple<K, V>> copyFailedRecords = new ArrayList<>(this.failedRecords);
-			this.failedRecords.clear();
+
+			// If we use failedRecords.clear() to remove copied record from failed records,
+			// We may encounter race condition during this operation.
+			// Other, the thread which execute this block, may miss one failed record.
+			int capturedRecordsCount = copyFailedRecords.size();
+			for (int i = 0; i < capturedRecordsCount; i++) {
+				this.failedRecords.pollFirst();
+			}
 
 			// If any copied and failed record fails to complete due to an unexpected error,
 			// We will give up on retrying with the remaining copied and failed Records.