diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java index bbf9fb351..0748488e4 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java @@ -67,7 +67,7 @@ public SmartCompositeMessageConverter(Collection converters, S public Object fromMessage(Message message, Class targetClass) { Collection 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 resultList = new ArrayList<>(); + List 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 iterator = getConverters().iterator(); iterator.hasNext() && !isConverted;) { + for (Iterator 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 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 snapshotConverters() { + return new ArrayList<>(getConverters()); + } } diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverterTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverterTests.java new file mode 100644 index 000000000..00bc5b52e --- /dev/null +++ b/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 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 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; + } + + } + +}