GH-2634: Reactor Kafka Binder Customization

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2634

Allow customization of `ReceiverOptions` and `ProducerOptions`.

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<ReceiverOptionsCustomizer> customizers) {
+		if (customizers.getIfUnique() != null) {
+			this.receiverOptionsCustomizer = customizers.getIfUnique();
+		}
+		else {
+			List<ReceiverOptionsCustomizer> list = customizers.orderedStream().toList();
+			ReceiverOptionsCustomizer customizer = (name, opts) -> {
+				ReceiverOptions<Object, Object> last = null;
+				for (ReceiverOptionsCustomizer cust: list) {
+					last = cust.apply(name, opts);
+				}
+				return last;
+			};
+			if (!list.isEmpty()) {
+				this.receiverOptionsCustomizer = customizer;
+			}
+		}
+	}
+
+	public void senderOptionsCustomizers(ObjectProvider<SenderOptionsCustomizer> customizers) {
+		if (customizers.getIfUnique() != null) {
+			this.senderOptionsCustomizer = customizers.getIfUnique();
+		}
+		else {
+			List<SenderOptionsCustomizer> list = customizers.orderedStream().toList();
+			SenderOptionsCustomizer customizer = (name, opts) -> {
+				SenderOptions<Object, Object> 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<KafkaProducerProperties> producerProperties, MessageChannel errorChannel)
@@ -113,7 +156,8 @@ public class ReactorKafkaBinder
 					destination.getName());
 		}
 
-		SenderOptions<Object, Object> opts = SenderOptions.create(configs);
+		SenderOptions<Object, Object> 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<Object, Object> opts = ReceiverOptions.create(configs)
 			.addAssignListener(parts -> logger.info("Assigned: " + parts))
 			.subscription(Collections.singletonList(destination.getName()));
+		opts = this.receiverOptionsCustomizer.apply(properties.getBindingName(), opts);
+		ReceiverOptions<Object, Object> 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));
 				}
 			}
 

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> consumerConfigCustomizer,
-			ObjectProvider<ProducerConfigCustomizer> producerConfigCustomizer) {
+			ObjectProvider<ProducerConfigCustomizer> producerConfigCustomizer,
+			ObjectProvider<ReceiverOptionsCustomizer> receiverOptionsCustomizers,
+			ObjectProvider<SenderOptionsCustomizer> 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;
 	}
 

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<String, ReceiverOptions<Object, Object>, ReceiverOptions<Object, Object>>, Ordered {
+
+	@Override
+	default int getOrder() {
+		return 0;
+	}
+
+}

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<String, SenderOptions<Object, Object>, SenderOptions<Object, Object>>, Ordered {
+
+	@Override
+	default int getOrder() {
+		return 0;
+	}
+
+}

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<String> recOptsCustOrder = new ArrayList<>();
+
 	@Autowired
 	private EmbeddedKafkaBroker embeddedKafka;
 
@@ -71,6 +76,7 @@ class ReactorKafkaBinderIntegrationTests {
 	@ValueSource(booleans = { false, true })
 	void endToEndReactorKafkaBinder(boolean excludeKafkaAutoConfig) {
 
+		recOptsCustOrder.clear();
 		Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("group1", "false", embeddedKafka);
 		consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 		DefaultKafkaConsumerFactory<String, String> 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<Object, Object> apply(String t, ReceiverOptions<Object, Object> u) {
+					recOptsCustOrder.add("two");
+					return u;
+				}
+
+				@Override
+				public int getOrder() {
+					return -1;
+				}
+
+
+			};
+		}
+
 	}
 
 }

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<SenderOptionsCustomizer> 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();
 	}
 
 }

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