GH-2994, GH-2986 Add initial support for customizing Message Converte…

…r behavior

primarily during batch processing.

binders/kafka-binder/spring-cloud-stream-binder-kafka/pom.xml

@@ -32,6 +32,11 @@
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-test-binder</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-autoconfigure</artifactId>

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/DefaultMessageConverterHelper.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2019-2024 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.stream.binder.kafka.config;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.springframework.cloud.function.context.config.MessageConverterHelper;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Oleg Zhurakousky
+ */
+public class DefaultMessageConverterHelper implements MessageConverterHelper {
+
+	@Override
+	public boolean shouldFailIfCantConvert(Message<?> message) {
+		return false;
+	}
+
+	public void postProcessBatchMessageOnFailure(Message<?> message, int index) {
+		AtomicInteger deliveryAttempt = (AtomicInteger) message.getHeaders().get("deliveryAttempt");
+//		if (message.getHeaders().containsKey("amqp_batchedHeaders") && deliveryAttempt != null && deliveryAttempt.get() == 1) {
+//			ArrayList<?> list = (ArrayList<?>) message.getHeaders().get("amqp_batchedHeaders");
+//			list.remove(index);
+//		}
+	}
+}

binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/FunctionBatchingConversionTests.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright 2019-2024 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.stream.binder.kafka;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.function.context.config.MessageConverterHelper;
+import org.springframework.cloud.function.json.JacksonMapper;
+import org.springframework.cloud.stream.binder.test.InputDestination;
+import org.springframework.cloud.stream.binder.test.OutputDestination;
+import org.springframework.cloud.stream.binder.test.TestChannelBinder;
+import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.converter.MessageConversionException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ *
+ */
+public class FunctionBatchingConversionTests {
+
+	@SuppressWarnings("unchecked")
+//	@Test
+	void testBatchHeadersMatchingPayload() {
+		TestChannelBinderConfiguration.applicationContextRunner(BatchFunctionConfiguration.class)
+				.withPropertyValues("spring.cloud.stream.function.definition=func",
+						"spring.cloud.stream.bindings.func-in-0.consumer.batch-mode=true",
+						"spring.cloud.stream.rabbit.bindings.func-in-0.consumer.enable-batching=true")
+				.run(context -> {
+					InputDestination inputDestination = context.getBean(InputDestination.class);
+					OutputDestination outputDestination = context.getBean(OutputDestination.class);
+
+					List<byte[]> payloads = List.of("hello".getBytes(StandardCharsets.UTF_8),
+							"{\"name\":\"Ricky\"}".getBytes(StandardCharsets.UTF_8),
+							"{\"name\":\"Julien\"}".getBytes(StandardCharsets.UTF_8),
+							"{\"name\":\"Bubbles\"}".getBytes(StandardCharsets.UTF_8),
+							"hello".getBytes(StandardCharsets.UTF_8));
+					List<Map<String, String>> amqpBatchHeaders = new ArrayList<>();
+					for (int i = 0; i < 5; i++) {
+						Map<String, String> batchHeaders = new LinkedHashMap<>();
+						batchHeaders.put("amqp_receivedDeliveryMode", "PERSISTENT");
+						batchHeaders.put("index", String.valueOf(i));
+						amqpBatchHeaders.add(batchHeaders);
+					}
+
+					var message = MessageBuilder.withPayload(payloads)
+							.setHeader("amqp_batchedHeaders", amqpBatchHeaders)
+							.setHeader("deliveryAttempt", new AtomicInteger(1)).build();
+					inputDestination.send(message);
+
+					Message<byte[]> resultMessage = outputDestination.receive();
+					JacksonMapper mapper = context.getBean(JacksonMapper.class);
+					List<?> resultPayloads = mapper.fromJson(resultMessage.getPayload(), List.class);
+					assertThat(resultPayloads).hasSize(3);
+
+					List<Map<String, String>> amqpBatchedHeaders = (List<Map<String, String>>) resultMessage
+							.getHeaders().get("amqp_batchedHeaders");
+					assertThat(amqpBatchedHeaders).hasSize(resultPayloads.size());
+					assertThat(amqpBatchedHeaders.get(0).get("index")).isEqualTo("1");
+					assertThat(amqpBatchedHeaders.get(1).get("index")).isEqualTo("2");
+					assertThat(amqpBatchedHeaders.get(2).get("index")).isEqualTo("3");
+
+					context.stop();
+				});
+	}
+
+//	@Test
+	void testBatchHeadersForcingFatalFailureOnConversiionException() {
+		TestChannelBinderConfiguration
+				.applicationContextRunner(BatchFunctionConfigurationWithAdditionalConversionHelper.class)
+				.withPropertyValues("spring.cloud.stream.function.definition=func",
+						"spring.cloud.stream.bindings.func-in-0.consumer.batch-mode=true",
+						"spring.cloud.stream.bindings.func-in-0.consumer.max-attempts=1",
+						"spring.cloud.stream.rabbit.bindings.func-in-0.consumer.enable-batching=true")
+				.run(context -> {
+					InputDestination inputDestination = context.getBean(InputDestination.class);
+
+					List<byte[]> payloads = List.of("hello".getBytes(StandardCharsets.UTF_8),
+							"{\"name\":\"Ricky\"}".getBytes(StandardCharsets.UTF_8),
+							"{\"name\":\"Julien\"}".getBytes(StandardCharsets.UTF_8),
+							"{\"name\":\"Bubbles\"}".getBytes(StandardCharsets.UTF_8),
+							"hello".getBytes(StandardCharsets.UTF_8));
+					List<Map<String, String>> amqpBatchHeaders = new ArrayList<>();
+					for (int i = 0; i < 5; i++) {
+						Map<String, String> batchHeaders = new LinkedHashMap<>();
+						batchHeaders.put("amqp_receivedDeliveryMode", "PERSISTENT");
+						batchHeaders.put("index", String.valueOf(i));
+						amqpBatchHeaders.add(batchHeaders);
+					}
+
+					var message = MessageBuilder.withPayload(payloads)
+							.setHeader("amqp_batchedHeaders", amqpBatchHeaders)
+							.setHeader("deliveryAttempt", new AtomicInteger(1)).build();
+					inputDestination.send(message);
+					TestChannelBinder binder = context.getBean(TestChannelBinder.class);
+					assertThat(binder.getLastError().getPayload()).isInstanceOf(MessageHandlingException.class);
+					MessageHandlingException exception = (MessageHandlingException) binder.getLastError().getPayload();
+					assertThat(exception.getCause()).isInstanceOf(MessageConversionException.class);
+
+					context.stop();
+				});
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	public static class BatchFunctionConfiguration {
+		@Bean
+		public Function<Message<List<Person>>, Message<List<Person>>> func() {
+			return x -> {
+				return x;
+			};
+		}
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	public static class BatchFunctionConfigurationWithAdditionalConversionHelper {
+
+		@Bean
+		public MessageConverterHelper helper() {
+			return new MessageConverterHelper() {
+				public boolean shouldFailIfCantConvert(Message<?> message) {
+					return true;
+				}
+			};
+		}
+
+		@Bean
+		public Function<Message<List<Person>>, Message<List<Person>>> func() {
+			return x -> {
+				return x;
+			};
+		}
+	}
+
+	static class Person {
+
+		private String name;
+
+		public String getName() {
+			return name;
+		}
+
+		public void setName(String name) {
+			this.name = name;
+		}
+
+		public String toString() {
+			return "name: " + name;
+		}
+	}
+}

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml

@@ -35,6 +35,11 @@
 			<artifactId>spring-boot-configuration-processor</artifactId>
 			<optional>true</optional>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-test-binder</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-stream</artifactId>

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/DefaultMessageConverterHelper.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2019-2024 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.stream.binder.rabbit.config;
+
+import java.util.ArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.springframework.cloud.function.context.config.MessageConverterHelper;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Oleg Zhurakousky
+ */
+public class DefaultMessageConverterHelper implements MessageConverterHelper {
+
+	@Override
+	public boolean shouldFailIfCantConvert(Message<?> message) {
+		return false;
+	}
+
+	public void postProcessBatchMessageOnFailure(Message<?> message, int index) {
+		AtomicInteger deliveryAttempt = (AtomicInteger) message.getHeaders().get("deliveryAttempt");
+		if (message.getHeaders().containsKey("amqp_batchedHeaders") && deliveryAttempt != null && deliveryAttempt.get() == 1) {
+			ArrayList<?> list = (ArrayList<?>) message.getHeaders().get("amqp_batchedHeaders");
+			list.remove(index);
+		}
+	}
+}

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/MessageConverterHelperConfiguration.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2019-2024 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.stream.binder.rabbit.config;
+
+import org.springframework.cloud.function.context.config.MessageConverterHelper;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Oleg Zhurakousky
+ */
+@Configuration(proxyBeanMethods = false)
+public class MessageConverterHelperConfiguration {
+
+	@Bean
+	public MessageConverterHelper messageConverterHelper() {
+		return new DefaultMessageConverterHelper();
+	}
+}

binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +1,2 @@
 org.springframework.cloud.stream.binder.rabbit.config.ExtendedBindingHandlerMappingsProviderConfiguration
+org.springframework.cloud.stream.binder.rabbit.config.MessageConverterHelperConfiguration