Skip to content

Commit

Permalink
GH-3726: Fix KafkaMessageListenerContainer for ConcurrentModification…
Browse files Browse the repository at this point in the history
…Exception

Fixes: #3726
Issue link: #3726

`KafkaMessageListenerContainer.getAssignedPartitions()` is not safe due to the fact that 
different threads can iterate/modify any of the fields `partitionsListenerConsumer.definedPartitions` or `partitionsListenerConsumer.assignedPartitions` simultaneously, 
but collection types of these fields are not designed for such scenarios. 
Thus at least `ConcurrentModificationException` can be thrown.

* Wrap `partitionsListenerConsumer.definedPartitions` and `partitionsListenerConsumer.assignedPartitions`  into `Collections.synchronizedSet()`

Signed-off-by: Tim Barabanov <[email protected]>

[[email protected] Fix commit message]

**Auto-cherry-pick to `3.2.x`**

Signed-off-by: Artem Bilan <[email protected]>
  • Loading branch information
tbarabanov authored Jan 30, 2025
1 parent 7aa87dd commit b772ebd
Showing 1 changed file with 5 additions and 3 deletions.
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright 2016-2024 the original author or authors.
* Copyright 2016-2025 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.
Expand Down Expand Up @@ -171,6 +171,7 @@
* @author Lokesh Alamuri
* @author Sanghyeok An
* @author Christian Fredriksson
* @author Timofey Barabanov
*/
public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
extends AbstractMessageListenerContainer<K, V> implements ConsumerPauseResumeEventPublisher {
Expand Down Expand Up @@ -625,7 +626,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume

private final Map<TopicPartition, Long> offsets = new LinkedHashMap<>();

private final Collection<TopicPartition> assignedPartitions = new LinkedHashSet<>();
private final Collection<TopicPartition> assignedPartitions = Collections.synchronizedSet(new LinkedHashSet<>());

private final Map<TopicPartition, OffsetAndMetadata> lastCommits = new HashMap<>();

Expand Down Expand Up @@ -1247,7 +1248,8 @@ private void subscribeOrAssignTopics(final Consumer<? super K, ? super V> subscr
else {
List<TopicPartitionOffset> topicPartitionsToAssign =
Arrays.asList(KafkaMessageListenerContainer.this.topicPartitions);
this.definedPartitions = new LinkedHashMap<>(topicPartitionsToAssign.size());
this.definedPartitions = Collections.synchronizedMap(
new LinkedHashMap<>(topicPartitionsToAssign.size()));
for (TopicPartitionOffset topicPartition : topicPartitionsToAssign) {
this.definedPartitions.put(topicPartition.getTopicPartition(),
new OffsetMetadata(topicPartition.getOffset(), topicPartition.isRelativeToCurrent(),
Expand Down

0 comments on commit b772ebd

Please sign in to comment.