smallrye#2756 Update the ConsumerBuilder and ProducerBuilder to confi…

…gure the (de)serializers before creating the KafkaConsumer/KafkaProducer instances.
diversit · Sep 16, 2024 · 1372825 · 1372825
1 parent c2ed366
commit 1372825
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/...mpanion/src/main/java/io/smallrye/reactive/messaging/kafka/companion/ConsumerBuilder.java b/...mpanion/src/main/java/io/smallrye/reactive/messaging/kafka/companion/ConsumerBuilder.java
@@ -143,6 +143,11 @@ public ConsumerBuilder(Map<String, Object> props, Duration kafkaApiTimeout,
             Deserializer<K> keyDeser, Deserializer<V> valueDeser) {
         this.props = props;
         this.kafkaApiTimeout = kafkaApiTimeout;
+
+        // configure deserializers
+        keyDeser.configure(props, true);
+        valueDeser.configure(props, false);
+
         this.consumerCreator = p -> new KafkaConsumer<>(p, keyDeser, valueDeser);
     }
 

diff --git a/...mpanion/src/main/java/io/smallrye/reactive/messaging/kafka/companion/ProducerBuilder.java b/...mpanion/src/main/java/io/smallrye/reactive/messaging/kafka/companion/ProducerBuilder.java
@@ -131,6 +131,11 @@ public ProducerBuilder(Map<String, Object> props, Duration kafkaApiTimeout,
             Serializer<K> keySerializer, Serializer<V> valueSerializer) {
         this.props = props;
         this.kafkaApiTimeout = kafkaApiTimeout;
+
+        // configure serializers
+        keySerializer.configure(props, true);
+        valueSerializer.configure(props, false);
+
         this.producerCreator = p -> new KafkaProducer<>(p, keySerializer, valueSerializer);
     }