spring-cloudGH-2685: Reactive Kafka Customizers - Add Generics

Resolves spring-cloud#2685

Required to allow customization of serializers/deserializers.

Author: garyrussell

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java

@@ -92,9 +92,9 @@ public class ReactorKafkaBinder
 
 	private ProducerConfigCustomizer producerConfigCustomizer;
 
-	private ReceiverOptionsCustomizer receiverOptionsCustomizer = (name, opts) -> opts;
+	private ReceiverOptionsCustomizer<Object, Object> receiverOptionsCustomizer = (name, opts) -> opts;
 
-	private SenderOptionsCustomizer senderOptionsCustomizer = (name, opts) -> opts;
+	private SenderOptionsCustomizer<Object, Object> senderOptionsCustomizer = (name, opts) -> opts;
 
 	public ReactorKafkaBinder(KafkaBinderConfigurationProperties configurationProperties,
 			KafkaTopicProvisioner provisioner) {
@@ -111,6 +111,7 @@ public void setProducerConfigCustomizer(ProducerConfigCustomizer producerConfigC
 		this.producerConfigCustomizer = producerConfigCustomizer;
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void receiverOptionsCustomizers(ObjectProvider<ReceiverOptionsCustomizer> customizers) {
 		if (customizers.getIfUnique() != null) {
 			this.receiverOptionsCustomizer = customizers.getIfUnique();
@@ -120,7 +121,7 @@ public void receiverOptionsCustomizers(ObjectProvider<ReceiverOptionsCustomizer>
 			ReceiverOptionsCustomizer customizer = (name, opts) -> {
 				ReceiverOptions<Object, Object> last = null;
 				for (ReceiverOptionsCustomizer cust: list) {
-					last = cust.apply(name, opts);
+					last = (ReceiverOptions<Object, Object>) cust.apply(name, opts);
 				}
 				return last;
 			};
@@ -130,6 +131,7 @@ public void receiverOptionsCustomizers(ObjectProvider<ReceiverOptionsCustomizer>
 		}
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void senderOptionsCustomizers(ObjectProvider<SenderOptionsCustomizer> customizers) {
 		if (customizers.getIfUnique() != null) {
 			this.senderOptionsCustomizer = customizers.getIfUnique();
@@ -139,7 +141,7 @@ public void senderOptionsCustomizers(ObjectProvider<SenderOptionsCustomizer> cus
 			SenderOptionsCustomizer customizer = (name, opts) -> {
 				SenderOptions<Object, Object> last = null;
 				for (SenderOptionsCustomizer cust: list) {
-					last = cust.apply(name, opts);
+					last = (SenderOptions<Object, Object>) cust.apply(name, opts);
 				}
 				return last;
 			};
@@ -161,7 +163,8 @@ protected MessageHandler createProducerMessageHandler(ProducerDestination destin
 					destination.getName());
 		}
 
-		SenderOptions<Object, Object> opts = this.senderOptionsCustomizer.apply(producerProperties.getBindingName(),
+		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();
@@ -188,7 +191,8 @@ protected MessageProducer createConsumerEndpoint(ConsumerDestination destination
 		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);
+		opts = this.receiverOptionsCustomizer.apply(properties.getBindingName(),
+				opts);
 		ReceiverOptions<Object, Object> finalOpts = opts;
 
 		class ReactorMessageProducer extends MessageProducerSupport {

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReceiverOptionsCustomizer.java

@@ -27,13 +27,18 @@
  * the second is the {@link ReceiverOptions} to customize. Return the customized
  * {@link ReceiverOptions}. Applied in {@link Ordered order} if multiple customizers are
  * found.
+ * <p>
+ * Generic since 4.0.3 to allow customization of deserializers.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
  *
  * @author Gary Russell
  * @since 4.0.2
  *
  */
-public interface ReceiverOptionsCustomizer
-		extends BiFunction<String, ReceiverOptions<Object, Object>, ReceiverOptions<Object, Object>>, Ordered {
+public interface ReceiverOptionsCustomizer<K, V>
+		extends BiFunction<String, ReceiverOptions<K, V>, ReceiverOptions<K, V>>, Ordered {
 
 	@Override
 	default int getOrder() {

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/SenderOptionsCustomizer.java

@@ -27,13 +27,18 @@
  * second is the {@link SenderOptions} to customize. Return the customized
  * {@link SenderOptions}. Applied in {@link Ordered order} if multiple customizers are
  * found.
+ * <p>
+ * Generic since 4.0.3 to allow customization of serializers.
+ *
+ * @param <K> the key type.
+ * @param <V> the value type.
  *
  * @author Gary Russell
  * @since 4.0.2
  *
  */
-public interface SenderOptionsCustomizer
-		extends BiFunction<String, SenderOptions<Object, Object>, SenderOptions<Object, Object>>, Ordered {
+public interface SenderOptionsCustomizer<K, V>
+		extends BiFunction<String, SenderOptions<K, V>, SenderOptions<K, V>>, Ordered {
 
 	@Override
 	default int getOrder() {

binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderIntegrationTests.java

@@ -26,6 +26,7 @@
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
@@ -167,20 +168,21 @@ public Function<Flux<ReceiverRecord<byte[], byte[]>>, Flux<String>> lowercase()
 		}
 
 		@Bean
-		ReceiverOptionsCustomizer cust1() {
+		ReceiverOptionsCustomizer<?, ?> cust1() {
 			return (t, u) -> {
 				recOptsCustOrder.add("one");
 				return u;
 			};
 		}
 
 		@Bean
-		ReceiverOptionsCustomizer cust2() {
-			return new ReceiverOptionsCustomizer() {
+		ReceiverOptionsCustomizer<String, byte[]> cust2() {
+			return new ReceiverOptionsCustomizer<>() {
 
 				@Override
-				public ReceiverOptions<Object, Object> apply(String t, ReceiverOptions<Object, Object> u) {
+				public ReceiverOptions<String, byte[]> apply(String t, ReceiverOptions<String, byte[]> u) {
 					recOptsCustOrder.add("two");
+					u.withKeyDeserializer(new StringDeserializer());
 					return u;
 				}
 
@@ -189,7 +191,6 @@ 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

@@ -220,6 +220,7 @@ void producerBinding() throws InterruptedException {
 		provisioner.setMetadataRetryOperations(new RetryTemplate());
 		ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner);
 		binder.setApplicationContext(new GenericApplicationContext());
+		@SuppressWarnings("rawtypes")
 		ObjectProvider<SenderOptionsCustomizer> cust = mock(ObjectProvider.class);
 		AtomicBoolean custCalled = new AtomicBoolean();
 		given(cust.getIfUnique()).willReturn((name, opts) -> {