diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/pom.xml b/binders/kafka-binder/spring-cloud-stream-binder-kafka/pom.xml index 040a9f28a..7302a3c86 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka/pom.xml +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka/pom.xml @@ -32,6 +32,11 @@ org.springframework.cloud spring-cloud-stream + + org.springframework.cloud + spring-cloud-stream-test-binder + test + org.springframework.boot spring-boot-autoconfigure diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/DefaultMessageConverterHelper.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/DefaultMessageConverterHelper.java new file mode 100644 index 000000000..631a13ac6 --- /dev/null +++ b/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); +// } + } +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/FunctionBatchingConversionTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/FunctionBatchingConversionTests.java new file mode 100644 index 000000000..a158bb8a0 --- /dev/null +++ b/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 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> amqpBatchHeaders = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + Map 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 resultMessage = outputDestination.receive(); + JacksonMapper mapper = context.getBean(JacksonMapper.class); + List resultPayloads = mapper.fromJson(resultMessage.getPayload(), List.class); + assertThat(resultPayloads).hasSize(3); + + List> amqpBatchedHeaders = (List>) 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 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> amqpBatchHeaders = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + Map 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>> 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>> 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; + } + } +} diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml index bfba6b864..263356905 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml @@ -35,6 +35,11 @@ spring-boot-configuration-processor true + + org.springframework.cloud + spring-cloud-stream-test-binder + test + org.springframework.cloud spring-cloud-stream diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/DefaultMessageConverterHelper.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/DefaultMessageConverterHelper.java new file mode 100644 index 000000000..2134c1438 --- /dev/null +++ b/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); + } + } +} diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/MessageConverterHelperConfiguration.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/MessageConverterHelperConfiguration.java new file mode 100644 index 000000000..ea1ca65ac --- /dev/null +++ b/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(); + } +} diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index a6aa77f21..dcf45ff3c 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/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 diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/FunctionBatchingConversionTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/FunctionBatchingConversionTests.java new file mode 100644 index 000000000..768b0a513 --- /dev/null +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/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.rabbit; + +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 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> amqpBatchHeaders = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + Map 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 resultMessage = outputDestination.receive(); + JacksonMapper mapper = context.getBean(JacksonMapper.class); + List resultPayloads = mapper.fromJson(resultMessage.getPayload(), List.class); + assertThat(resultPayloads).hasSize(3); + + List> amqpBatchedHeaders = (List>) 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 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> amqpBatchHeaders = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + Map 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>> 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>> 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; + } + } +} diff --git a/schema-registry/spring-cloud-stream-schema-registry-client/pom.xml b/schema-registry/spring-cloud-stream-schema-registry-client/pom.xml index 61a7d42de..d9493ac5d 100644 --- a/schema-registry/spring-cloud-stream-schema-registry-client/pom.xml +++ b/schema-registry/spring-cloud-stream-schema-registry-client/pom.xml @@ -72,15 +72,19 @@ test - - - - - - + + org.apache.maven.plugins + maven-checkstyle-plugin + + + checkstyle-validation + none + + + org.apache.avro avro-maven-plugin