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 695200787..d41ef2e98 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
@@ -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 class ReactorKafkaBinder
 		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 class ReactorKafkaBinder
 			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 class ReactorKafkaBinder
 		}
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
 	public void senderOptionsCustomizers(ObjectProvider<SenderOptionsCustomizer> customizers) {
 		if (customizers.getIfUnique() != null) {
 			this.senderOptionsCustomizer = customizers.getIfUnique();
@@ -139,7 +141,7 @@ public class ReactorKafkaBinder
 			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;
 			};
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
index f00fabe71..2f2fbc230 100644
--- 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
@@ -27,13 +27,18 @@ import org.springframework.core.Ordered;
  * 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() {
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
index 0f43af6a4..37070e865 100644
--- 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
@@ -27,13 +27,18 @@ import org.springframework.core.Ordered;
  * 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() {
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 4db2d0011..6be427eee 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
@@ -26,6 +26,7 @@ import org.apache.kafka.clients.consumer.Consumer;
 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,7 +168,7 @@ class ReactorKafkaBinderIntegrationTests {
 		}
 
 		@Bean
-		ReceiverOptionsCustomizer cust1() {
+		ReceiverOptionsCustomizer<?, ?> cust1() {
 			return (t, u) -> {
 				recOptsCustOrder.add("one");
 				return u;
@@ -175,12 +176,13 @@ class ReactorKafkaBinderIntegrationTests {
 		}
 
 		@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 @@ class ReactorKafkaBinderIntegrationTests {
 					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 3bff3364d..a1e2a2c69 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
@@ -220,6 +220,7 @@ public class ReactorKafkaBinderTests {
 		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) -> {