Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3764: Replace LinkedList with ArrayList in listener container for records

Browse Source 
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]
Signed-off-by: Artem Bilan <artem.bilan@broadcom.com>
(cherry picked from commit 53149d4e65)
This commit is contained in:
Janek Lasocki-Biczysko
2025-02-24 20:15:40 +00:00
committed by Spring Builds
parent a07a55d19a
commit 051a8cb440
1 changed files with 4 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;
}
/**