Skip to content

Commit

Permalink
GH-3726: Avoid ConcurrentModificationException in KafkaMessageListene…
Browse files Browse the repository at this point in the history
…rContainer

Fixes GH-3726 (#3726)

Signed-off-by: Tim Barabanov <[email protected]>
  • Loading branch information
tbarabanov committed Jan 30, 2025
1 parent 7aa87dd commit d3a4ae7
Showing 1 changed file with 4 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 @@ -625,7 +625,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 +1247,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 d3a4ae7

Please sign in to comment.