From a325b8b9fd41202d3251330e5fc34958b2098dfb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 26 Jan 2023 11:21:08 -0500 Subject: [PATCH] GH-2634: Reactor Kafka Binder Customization Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2634 Allow customization of `ReceiverOptions` and `ProducerOptions`. --- .../reactorkafka/ReactorKafkaBinder.java | 50 ++++++++++++++++++- .../ReactorKafkaBinderConfiguration.java | 8 ++- .../ReceiverOptionsCustomizer.java | 43 ++++++++++++++++ .../reactorkafka/SenderOptionsCustomizer.java | 43 ++++++++++++++++ .../ReactorKafkaBinderIntegrationTests.java | 37 +++++++++++++- .../reactorkafka/ReactorKafkaBinderTests.java | 14 +++++- .../main/asciidoc/kafka/kafka-reactive.adoc | 6 ++- 7 files changed, 194 insertions(+), 7 deletions(-) create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java create mode 100644 binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index e46553710..736ef26a3 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java @@ -36,6 +36,7 @@ import reactor.kafka.sender.SenderOptions; import reactor.kafka.sender.SenderRecord; import reactor.kafka.sender.SenderResult; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; @@ -86,6 +87,10 @@ public class ReactorKafkaBinder private ProducerConfigCustomizer producerConfigCustomizer; + private ReceiverOptionsCustomizer receiverOptionsCustomizer = (name, opts) -> opts; + + private SenderOptionsCustomizer senderOptionsCustomizer = (name, opts) -> opts; + public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties, KafkaTopicProvisioner provisioner) { @@ -101,6 +106,44 @@ public class ReactorKafkaBinder this.producerConfigCustomizer = producerConfigCustomizer; } + public void receiverOptionsCustomizers(ObjectProvider customizers) { + if (customizers.getIfUnique() != null) { + this.receiverOptionsCustomizer = customizers.getIfUnique(); + } + else { + List list = customizers.orderedStream().toList(); + ReceiverOptionsCustomizer customizer = (name, opts) -> { + ReceiverOptions last = null; + for (ReceiverOptionsCustomizer cust: list) { + last = cust.apply(name, opts); + } + return last; + }; + if (!list.isEmpty()) { + this.receiverOptionsCustomizer = customizer; + } + } + } + + public void senderOptionsCustomizers(ObjectProvider customizers) { + if (customizers.getIfUnique() != null) { + this.senderOptionsCustomizer = customizers.getIfUnique(); + } + else { + List list = customizers.orderedStream().toList(); + SenderOptionsCustomizer customizer = (name, opts) -> { + SenderOptions last = null; + for (SenderOptionsCustomizer cust: list) { + last = cust.apply(name, opts); + } + return last; + }; + if (!list.isEmpty()) { + this.senderOptionsCustomizer = customizer; + } + } + } + @Override protected MessageHandler createProducerMessageHandler(ProducerDestination destination, ExtendedProducerProperties producerProperties, MessageChannel errorChannel) @@ -113,7 +156,8 @@ public class ReactorKafkaBinder destination.getName()); } - SenderOptions opts = SenderOptions.create(configs); + SenderOptions opts = this.senderOptionsCustomizer.apply(producerProperties.getBindingName(), + SenderOptions.create(configs)); // TODO bean for converter; MCB doesn't use one on the producer side. RecordMessageConverter converter = new MessagingMessageConverter(); return new ReactorMessageHandler(opts, converter, destination.getName()); @@ -139,6 +183,8 @@ public class ReactorKafkaBinder ReceiverOptions opts = ReceiverOptions.create(configs) .addAssignListener(parts -> logger.info("Assigned: " + parts)) .subscription(Collections.singletonList(destination.getName())); + opts = this.receiverOptionsCustomizer.apply(properties.getBindingName(), opts); + ReceiverOptions finalOpts = opts; class ReactorMessageProducer extends MessageProducerSupport { @@ -146,7 +192,7 @@ public class ReactorKafkaBinder ReactorMessageProducer() { for (int i = 0; i < properties.getConcurrency(); i++) { - this.receivers.add(KafkaReceiver.create(opts)); + this.receivers.add(KafkaReceiver.create(finalOpts)); } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java index ceb2547ee..3d310fb4b 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2022 the original author or authors. + * Copyright 2022-2023 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. @@ -60,12 +60,16 @@ public class ReactorKafkaBinderConfiguration { KafkaTopicProvisioner provisioningProvider, KafkaExtendedBindingProperties extendedBindingProperties, ObjectProvider consumerConfigCustomizer, - ObjectProvider producerConfigCustomizer) { + ObjectProvider producerConfigCustomizer, + ObjectProvider receiverOptionsCustomizers, + ObjectProvider senderOptionsptionsCustomizers) { ReactorKafkaBinder reactorKafkaBinder = new ReactorKafkaBinder(configurationProperties, provisioningProvider); reactorKafkaBinder.setExtendedBindingProperties(extendedBindingProperties); reactorKafkaBinder.setConsumerConfigCustomizer(consumerConfigCustomizer.getIfUnique()); reactorKafkaBinder.setProducerConfigCustomizer(producerConfigCustomizer.getIfUnique()); + reactorKafkaBinder.receiverOptionsCustomizers(receiverOptionsCustomizers); + reactorKafkaBinder.senderOptionsCustomizers(senderOptionsptionsCustomizers); return reactorKafkaBinder; } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java new file mode 100644 index 000000000..022af1953 --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java @@ -0,0 +1,43 @@ +/* + * Copyright 2023-2023 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.reactorkafka; + +import java.util.function.BiFunction; + +import reactor.kafka.receiver.ReceiverOptions; + +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; + +/** + * Customizer for {@link ReceiverOptions}; the first parameter contains the binder name; + * the second is the {@link ReceiverOptions} to customize. Return the customized + * {@link ReceiverOptions}. Applied in {@link Order} if multiple customizers are found. + * + * @author Gary Russell + * @since 4.0.2 + * + */ +public interface ReceiverOptionsCustomizer + extends BiFunction, ReceiverOptions>, Ordered { + + @Override + default int getOrder() { + return 0; + } + +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java new file mode 100644 index 000000000..52ff2f59d --- /dev/null +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java @@ -0,0 +1,43 @@ +/* + * Copyright 2023-2023 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.reactorkafka; + +import java.util.function.BiFunction; + +import reactor.kafka.sender.SenderOptions; + +import org.springframework.core.Ordered; +import org.springframework.core.annotation.Order; + +/** + * Customizer for {@link SenderOptions}; the first parameter contains the binder name; the + * second is the {@link SenderOptions} to customize. Return the customized + * {@link SenderOptions}. Applied in {@link Order} if multiple customizers are found. + * + * @author Gary Russell + * @since 4.0.2 + * + */ +public interface SenderOptionsCustomizer + extends BiFunction, SenderOptions>, Ordered { + + @Override + default int getOrder() { + return 0; + } + +} diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java index 9c6d90454..4db2d0011 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022-2022 the original author or authors. + * Copyright 2022-2023 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. @@ -17,6 +17,8 @@ package org.springframework.cloud.stream.binder.reactorkafka; import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.function.Function; @@ -28,6 +30,7 @@ import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import reactor.core.publisher.Flux; +import reactor.kafka.receiver.ReceiverOptions; import reactor.kafka.receiver.ReceiverRecord; import org.springframework.beans.factory.annotation.Autowired; @@ -64,6 +67,8 @@ import static org.assertj.core.api.Assertions.assertThat; @EmbeddedKafka(topics = { "uppercased-words", "lowercased-words" }) class ReactorKafkaBinderIntegrationTests { + private static final List recOptsCustOrder = new ArrayList<>(); + @Autowired private EmbeddedKafkaBroker embeddedKafka; @@ -71,6 +76,7 @@ class ReactorKafkaBinderIntegrationTests { @ValueSource(booleans = { false, true }) void endToEndReactorKafkaBinder(boolean excludeKafkaAutoConfig) { + recOptsCustOrder.clear(); Map consumerProps = KafkaTestUtils.consumerProps("group1", "false", embeddedKafka); consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); @@ -112,6 +118,8 @@ class ReactorKafkaBinderIntegrationTests { .isNotNull() .extracting(ConsumerRecord::value) .isEqualTo("bazqux"); + + assertThat(recOptsCustOrder).containsExactly("two", "one", "two", "one"); } finally { pf.destroy(); @@ -158,6 +166,33 @@ class ReactorKafkaBinderIntegrationTests { return s -> s.map(rec -> new String(rec.value()).toLowerCase()); } + @Bean + ReceiverOptionsCustomizer cust1() { + return (t, u) -> { + recOptsCustOrder.add("one"); + return u; + }; + } + + @Bean + ReceiverOptionsCustomizer cust2() { + return new ReceiverOptionsCustomizer() { + + @Override + public ReceiverOptions apply(String t, ReceiverOptions u) { + recOptsCustOrder.add("two"); + return u; + } + + @Override + public int getOrder() { + return -1; + } + + + }; + } + } } diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java index 53c5af4d2..4ab3b5afe 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2021-2022 the original author or authors. + * Copyright 2021-2023 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. @@ -21,6 +21,7 @@ import java.util.HashSet; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.producer.RecordMetadata; @@ -29,6 +30,7 @@ import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import reactor.core.publisher.Mono; +import org.springframework.beans.factory.ObjectProvider; import org.springframework.boot.autoconfigure.kafka.KafkaProperties; import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; @@ -51,6 +53,7 @@ import org.springframework.messaging.MessageChannel; import org.springframework.retry.support.RetryTemplate; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; /** @@ -197,6 +200,13 @@ public class ReactorKafkaBinderTests { provisioner.setMetadataRetryOperations(new RetryTemplate()); ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner); binder.setApplicationContext(new GenericApplicationContext()); + ObjectProvider cust = mock(ObjectProvider.class); + AtomicBoolean custCalled = new AtomicBoolean(); + given(cust.getIfUnique()).willReturn((name, opts) -> { + custCalled.set(true); + return opts; + }); + binder.senderOptionsCustomizers(cust); MessageChannel outbound = new FluxMessageChannel(); KafkaProducerProperties ext = new KafkaProducerProperties(); @@ -213,6 +223,8 @@ public class ReactorKafkaBinderTests { latch.countDown(); }).subscribe(); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + bindProducer.unbind(); + assertThat(custCalled).isTrue(); } } diff --git a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc b/docs/src/main/asciidoc/kafka/kafka-reactive.adoc index ff5763a8c..24a4b6849 100644 --- a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc +++ b/docs/src/main/asciidoc/kafka/kafka-reactive.adoc @@ -40,7 +40,11 @@ This is because the underlying binder is built on top of https://projectreactor. On the consumer side, it makes use of the https://projectreactor.io/docs/kafka/release/reference/#api-guide-receiver[KafkaReceiver] which is a reactive implementation of a Kafka consumer. Similarly, on the producer side, it uses https://projectreactor.io/docs/kafka/release/reference/#api-guide-sender[KafkaSender] API which is the reactive implementation of a Kafka producer. Since the foundations of the reactive Kafka binder is built upon a proper reactive Kafka API, applications get the full benefits of using reactive technologies. -Things like automatic backpressure among other reactive capabilities are built-in for the application when using this reactive Kafka binder. +Things like automatic back pressure, among other reactive capabilities, are built-in for the application when using this reactive Kafka binder. + +Starting with version 4.0.2, you can customize the `ReceiverOptions` and `SenderOptions` by providing one or more `ReceiverOptionsCustomizer` or `SenderOptionsCustomizer` beans respectively. +They are `BiFunction` s which receive the binding name and initial options, returning the customized options. +The interfaces extend `Ordered` so the customizers will be applied in the order required, when more than one are present. === Consuming Records in the Raw Format