diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
index 2f0c82600..5d368d75c 100644
--- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
+++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java
@@ -20,10 +20,11 @@ import java.io.IOException;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.stream.Collectors;
 
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -45,6 +46,7 @@ import org.springframework.messaging.converter.MessageConversionException;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @since 2.0
+ * @deprecated since 3.2 as we are no longer needed since functional-based programming model is no longer using it.
  */
 class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter {
 
@@ -139,27 +141,25 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
 			else {
 				final JavaType typeToUse = type;
 				if (payload instanceof Collection) {
-					Collection<?> collection = ((Collection<?>) payload).stream()
-							.map(value -> {
-								try {
-									if (value instanceof byte[]) {
-										return objectMapper.readValue((byte[]) value, typeToUse.getContentType());
-									}
-									else if (value instanceof String) {
-										return objectMapper.readValue((String) value, typeToUse.getContentType());
-									}
-									else {
-										// fall back to simple type-conversion
-										// see https://github.com/spring-cloud/spring-cloud-stream/issues/1898
-										return objectMapper.convertValue(value, typeToUse.getContentType());
-									}
-								}
-								catch (Exception e) {
-									logger.error("Failed to convert payload " + value, e);
-								}
-								return null;
-							}).collect(Collectors.toList());
-
+					List<Object> collection = new ArrayList<>();
+					for (Object value : ((Collection<?>) payload)) {
+						try {
+							if (value instanceof byte[]) {
+								collection.add(objectMapper.readValue((byte[]) value, typeToUse.getContentType()));
+							}
+							else if (value instanceof String) {
+								collection.add(objectMapper.readValue((String) value, typeToUse.getContentType()));
+							}
+							else {
+								// fall back to simple type-conversion
+								// see https://github.com/spring-cloud/spring-cloud-stream/issues/1898
+								collection.add(objectMapper.convertValue(value, typeToUse.getContentType()));
+							}
+						}
+						catch (Exception e) {
+							throw new MessageConversionException("Failed to convert payload " + value, e);
+						}
+					}
 					return collection;
 				}
 				return null;
diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java
index 468b3395d..b62dc930c 100644
--- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java
+++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverterTests.java
@@ -16,19 +16,25 @@
 
 package org.springframework.cloud.stream.converter;
 
+import java.util.Collections;
+import java.util.Map;
+
 import com.fasterxml.jackson.databind.DeserializationFeature;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.MapperFeature;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.json.JsonMapper;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.core.ResolvableType;
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.support.GenericMessage;
 
 import static org.junit.Assert.fail;
 
-
+@SuppressWarnings("deprecation")
 public class ApplicationJsonMessageMarshallingConverterTests {
+
 	@Test
 	void badJson() {
 
@@ -47,10 +53,29 @@ public class ApplicationJsonMessageMarshallingConverterTests {
 		}
 	}
 
+	@Test
+	void errorPropagationTestOnCollection() {
+		ApplicationJsonMessageMarshallingConverter converter = new ApplicationJsonMessageMarshallingConverter(
+				JsonMapper.builder().build());
+
+		try {
+			converter.fromMessage(new GenericMessage<>(Collections.singletonList("{ \"field1\": 1 }")), Map.class,
+					ResolvableType.forClassWithGenerics(Map.class, String.class, String.class).getType());
+			fail();
+		}
+		catch (MessageConversionException e) {
+			// good
+		}
+		catch (Throwable t) {
+			fail();
+		}
+	}
+
 	private ObjectMapper initObjectMapper() {
 		ObjectMapper objectMapper = new ObjectMapper();
 		objectMapper.configure(MapperFeature.DEFAULT_VIEW_INCLUSION, false);
 		objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
 		return objectMapper;
 	}
+
 }