From 53149d4e6557420172b410fea8a12b10bdf320ee Mon Sep 17 00:00:00 2001
From: Janek Lasocki-Biczysko <janek.lb@gmail.com>
Date: Mon, 24 Feb 2025 20:15:40 +0000
Subject: [PATCH] GH-3764: Replace LinkedList with ArrayList in listener
 container for records

Fixes: #3764
Issue link: https://github.com/spring-projects/spring-kafka/issues/3764

Acknowledging an index in a batch has quadratic time `N(N+1)/2` ~ `N^2`

Batch consumers operate on a `LinkedList` of records.
If the consumer uses `MANUAL_IMMEDIATE` ack mode, and the listener invokes `acknowledgement.acknowledge(index)` where index is relatively big (e.g. when processing batches of `100k`), performance takes hit because of the linear lookup `records.get(i)` in a loop

Signed-off-by: Janek Lasocki-Biczysko <janek.lb@gmail.com>

[artem.bilan@broadcom.com: improve commit message]
**Auto-cherry-pick to `3.3.x`**
Signed-off-by: Artem Bilan <artem.bilan@broadcom.com>
---
 .../kafka/listener/KafkaMessageListenerContainer.java  | 10 ++++------
 1 file changed, 4 insertions(+), 6 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
index a859cce0..87222dfe 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
@@ -172,6 +172,7 @@ import org.springframework.util.StringUtils;
  * @author Sanghyeok An
  * @author Christian Fredriksson
  * @author Timofey Barabanov
+ * @author Janek Lasocki-Biczysko
  */
 public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		extends AbstractMessageListenerContainer<K, V> implements ConsumerPauseResumeEventPublisher {
@@ -2237,12 +2238,9 @@ public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		}
 
 		private List<ConsumerRecord<K, V>> createRecordList(final ConsumerRecords<K, V> records) {
-			Iterator<ConsumerRecord<K, V>> iterator = records.iterator();
-			List<ConsumerRecord<K, V>> list = new LinkedList<>();
-			while (iterator.hasNext()) {
-				list.add(iterator.next());
-			}
-			return list;
+			List<ConsumerRecord<K, V>> recordList = new ArrayList<>(records.count());
+			records.forEach(recordList::add);
+			return recordList;
 		}
 
 		/**