GH-3726: Fix KafkaMessageListenerContainer for ConcurrentModificationException

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]

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

Signed-off-by: Artem Bilan <[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.
@@ -164,6 +164,7 @@
  * @author Raphael Rösch
  * @author Christian Mergenthaler
  * @author Mikael Carlstedt
+ * @author Timofey Barabanov
  */
 public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		extends AbstractMessageListenerContainer<K, V> implements ConsumerPauseResumeEventPublisher {
@@ -617,7 +618,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<>();
 
@@ -1219,7 +1220,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(),