fix(context): snapshot message converters to prevent ConcurrentModificationException

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java

@@ -67,7 +67,7 @@ public SmartCompositeMessageConverter(Collection<MessageConverter> converters, S
 	public Object fromMessage(Message<?> message, Class<?> targetClass) {
 		Collection<MessageConverterHelper> messageConverterHelpers = this.messageConverterHelpersSupplier != null
 				? this.messageConverterHelpersSupplier.get() : Collections.emptyList();
-		for (MessageConverter converter : getConverters()) {
+		for (MessageConverter converter : snapshotConverters()) {
 			if (!(message.getPayload() instanceof byte[]) && targetClass.isInstance(message.getPayload()) && !(message.getPayload() instanceof Collection<?>)) {
 				return message.getPayload();
 			}
@@ -102,13 +102,14 @@ public Object fromMessage(Message<?> message, Class<?> targetClass, @Nullable Ob
 			Type genericItemType = FunctionTypeUtils.getImmediateGenericType((Type) conversionHint, 0);
 			Class<?> genericItemRawType = FunctionTypeUtils.getRawType(genericItemType);
 			List<Object> resultList = new ArrayList<>();
+			List<MessageConverter> convertersSnapshot = snapshotConverters();
 			for (Object item : iterablePayload) {
 				boolean isConverted = false;
 				if (item.getClass().getName().startsWith("org.springframework.kafka.support.KafkaNull")) {
 					resultList.add(null);
 					isConverted = true;
 				}
-				for (Iterator<MessageConverter> iterator = getConverters().iterator(); iterator.hasNext() && !isConverted;) {
+				for (Iterator<MessageConverter> iterator = convertersSnapshot.iterator(); iterator.hasNext() && !isConverted;) {
 					MessageConverter converter = (MessageConverter) iterator.next();
 					if (!converter.getClass().getName().endsWith("ApplicationJsonMessageMarshallingConverter")) { // TODO Stream stuff, needs to be removed
 						Message<?> m  = MessageBuilder.withPayload(item).copyHeaders(message.getHeaders()).build(); // TODO Message creating may be expensive
@@ -129,7 +130,7 @@ public Object fromMessage(Message<?> message, Class<?> targetClass, @Nullable Ob
 			return resultList;
 		}
 		else {
-			for (MessageConverter converter : getConverters()) {
+			for (MessageConverter converter : snapshotConverters()) {
 				if (!converter.getClass().getName().endsWith("ApplicationJsonMessageMarshallingConverter")) { // TODO Stream stuff, needs to be removed
 					result = (converter instanceof SmartMessageConverter ?
 							((SmartMessageConverter) converter).fromMessage(message, targetClass, conversionHint) :
@@ -162,7 +163,7 @@ private void postProcessBatchMessage(Message<?> message, Collection<MessageConve
 	@Override
 	@Nullable
 	public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers) {
-		for (MessageConverter converter : getConverters()) {
+		for (MessageConverter converter : snapshotConverters()) {
 			if (headers.get(MessageHeaders.CONTENT_TYPE) == null) {
 				return null;
 			}
@@ -203,7 +204,7 @@ public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers) {
 	@Override
 	@Nullable
 	public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {
-		for (MessageConverter converter : getConverters()) {
+		for (MessageConverter converter : snapshotConverters()) {
 			Object value = headers.get(MessageHeaders.CONTENT_TYPE).toString();
 			String[] contentTypes = StringUtils.delimitedListToStringArray((String) value, ",");
 			for (String contentType : contentTypes) {
@@ -232,4 +233,8 @@ public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers, @N
 		}
 		return null;
 	}
+
+	private List<MessageConverter> snapshotConverters() {
+		return new ArrayList<>(getConverters());
+	}
 }

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverterTests.java

@@ -0,0 +1,161 @@
+/*
+ * Copyright 2015-present 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.function.context.config;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.jspecify.annotations.NullMarked;
+import org.jspecify.annotations.Nullable;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.util.MimeTypeUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link SmartCompositeMessageConverter} guarding against
+ * {@link java.util.ConcurrentModificationException} (see GH-1442) when the underlying
+ * converters list is structurally modified while a conversion is being performed.
+ */
+class SmartCompositeMessageConverterTests {
+
+	@Test
+	void toMessageSnapshotsConvertersWhenListMutatedDuringIteration() {
+		SmartCompositeMessageConverter composite = new SmartCompositeMessageConverter(
+				List.of(new NoOpMessageConverter()));
+		composite.getConverters().add(new MutatingMessageConverter(composite));
+		MessageHeaders headers = new MessageHeaders(
+				Map.of(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON));
+
+		assertThat(composite.toMessage("payload", headers)).isNull();
+	}
+
+	@Test
+	void fromMessageSnapshotsConvertersWhenListMutatedDuringIteration() {
+		SmartCompositeMessageConverter composite = new SmartCompositeMessageConverter(
+				List.of(new NoOpMessageConverter()));
+		composite.getConverters().add(new MutatingMessageConverter(composite));
+		Message<String> message = MessageBuilder.withPayload("hello").build();
+
+		assertThat(composite.fromMessage(message, Integer.class)).isNull();
+	}
+
+	@Test
+	void toMessageWithCrossThreadListMutationDuringIterationIsSafe() throws Exception {
+		SmartCompositeMessageConverter composite = new SmartCompositeMessageConverter(
+				List.of(new NoOpMessageConverter()));
+		CountDownLatch conversionEntered = new CountDownLatch(1);
+		CountDownLatch releaseConversion = new CountDownLatch(1);
+		composite.getConverters().add(new BlockingMessageConverter(conversionEntered, releaseConversion));
+		MessageHeaders headers = new MessageHeaders(
+				Map.of(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON));
+		AtomicReference<Throwable> failure = new AtomicReference<>();
+
+		Thread reader = new Thread(() -> {
+			try {
+				composite.toMessage("payload", headers);
+			}
+			catch (Throwable t) {
+				failure.set(t);
+			}
+		});
+		reader.start();
+		try {
+			// Wait until the reader is suspended inside the converters iteration,
+			// then structurally modify the live list from this thread.
+			assertThat(conversionEntered.await(5, TimeUnit.SECONDS)).isTrue();
+			composite.getConverters().add(new NoOpMessageConverter());
+		}
+		finally {
+			releaseConversion.countDown();
+		}
+		reader.join(5000);
+
+		assertThat(reader.isAlive()).isFalse();
+		assertThat(failure.get()).isNull();
+	}
+
+	@NullMarked
+	private record MutatingMessageConverter(SmartCompositeMessageConverter composite) implements MessageConverter {
+
+		@Override
+		@Nullable
+		public Object fromMessage(Message<?> message, Class<?> targetClass) {
+			this.composite().getConverters().add(new MutatingMessageConverter(this.composite()));
+			return null;
+		}
+
+		@Override
+		@Nullable
+		public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers) {
+			this.composite().getConverters().add(new MutatingMessageConverter(this.composite()));
+			return null;
+		}
+
+	}
+
+	@NullMarked
+	private record BlockingMessageConverter(CountDownLatch conversionEntered, CountDownLatch releaseConversion)
+			implements MessageConverter {
+
+		@Override
+		@Nullable
+		public Object fromMessage(Message<?> message, Class<?> targetClass) {
+			return null;
+		}
+
+		@Override
+		@Nullable
+		public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers) {
+			this.conversionEntered().countDown();
+			try {
+				this.releaseConversion().await();
+			}
+			catch (InterruptedException ex) {
+				Thread.currentThread().interrupt();
+			}
+			return null;
+		}
+
+	}
+
+	@NullMarked
+	private static final class NoOpMessageConverter implements MessageConverter {
+
+		@Override
+		@Nullable
+		public Object fromMessage(Message<?> message, Class<?> targetClass) {
+			return null;
+		}
+
+		@Override
+		@Nullable
+		public Message<?> toMessage(Object payload, @Nullable MessageHeaders headers) {
+			return null;
+		}
+
+	}
+
+}