GH-3726: Avoid ConcurrentModificationException in KafkaMessageListenerContainer

Fixes GH-3726 (#3726)

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

Author: tbarabanov

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

@@ -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.
@@ -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<>();
 
@@ -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(),