Retrieve kafka config in places where Override connector config is used

Author: ozangunalp

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/fault/KafkaDeadLetterQueue.java

@@ -36,6 +36,7 @@
 import io.smallrye.reactive.messaging.kafka.KafkaProducer;
 import io.smallrye.reactive.messaging.kafka.SerializationFailureHandler;
 import io.smallrye.reactive.messaging.kafka.api.OutgoingKafkaRecordMetadata;
+import io.smallrye.reactive.messaging.kafka.impl.ConfigHelper;
 import io.smallrye.reactive.messaging.kafka.impl.ReactiveKafkaProducer;
 import io.smallrye.reactive.messaging.providers.impl.ConnectorConfig;
 import io.smallrye.reactive.messaging.providers.impl.OverrideConnectorConfig;
@@ -84,6 +85,10 @@ public static class Factory implements KafkaFailureHandler.Factory {
         @Inject
         Instance<Config> rootConfig;
 
+        @Inject
+        @Any
+        Instance<Map<String, Object>> configurations;
+
         @Override
         public KafkaFailureHandler create(KafkaConnectorIncomingConfiguration config,
                 Vertx vertx,
@@ -104,7 +109,8 @@ public KafkaFailureHandler create(KafkaConnectorIncomingConfiguration config,
                             "key-serialization-failure-handler", c -> "dlq-serialization",
                             "value-serialization-failure-handler", c -> "dlq-serialization",
                             INTERCEPTOR_CLASSES_CONFIG, c -> ""));
-            KafkaConnectorOutgoingConfiguration producerConfig = new KafkaConnectorOutgoingConfiguration(connectorConfig);
+            Config kafkaConfig = ConfigHelper.retrieveChannelConfiguration(configurations, connectorConfig);
+            KafkaConnectorOutgoingConfiguration producerConfig = new KafkaConnectorOutgoingConfiguration(kafkaConfig);
 
             String deadQueueTopic = config.getDeadLetterQueueTopic().orElse("dead-letter-topic-" + config.getChannel());
 

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/fault/KafkaDelayedRetryTopic.java

@@ -54,6 +54,7 @@
 import io.smallrye.reactive.messaging.kafka.api.OutgoingKafkaRecordMetadata;
 import io.smallrye.reactive.messaging.kafka.commit.ContextHolder;
 import io.smallrye.reactive.messaging.kafka.commit.KafkaLatestCommit;
+import io.smallrye.reactive.messaging.kafka.impl.ConfigHelper;
 import io.smallrye.reactive.messaging.kafka.impl.ReactiveKafkaConsumer;
 import io.smallrye.reactive.messaging.kafka.impl.ReactiveKafkaProducer;
 import io.smallrye.reactive.messaging.providers.impl.ConnectorConfig;
@@ -101,6 +102,10 @@ public static class Factory implements KafkaFailureHandler.Factory {
         @Inject
         Instance<Config> rootConfig;
 
+        @Inject
+        @Any
+        Instance<Map<String, Object>> configurations;
+
         @Override
         public KafkaFailureHandler create(KafkaConnectorIncomingConfiguration config,
                 Vertx vertx,
@@ -131,7 +136,8 @@ public KafkaFailureHandler create(KafkaConnectorIncomingConfiguration config,
                             "key-serialization-failure-handler", c -> "dlq-serialization",
                             "value-serialization-failure-handler", c -> "dlq-serialization",
                             INTERCEPTOR_CLASSES_CONFIG, c -> ""));
-            KafkaConnectorOutgoingConfiguration producerConfig = new KafkaConnectorOutgoingConfiguration(connectorConfig);
+            Config kafkaConfig = ConfigHelper.retrieveChannelConfiguration(configurations, connectorConfig);
+            KafkaConnectorOutgoingConfiguration producerConfig = new KafkaConnectorOutgoingConfiguration(kafkaConfig);
 
             log.delayedRetryTopic(config.getChannel(), retryTopics, maxRetries, retryTimeout, deadQueueTopic);
 
@@ -144,7 +150,8 @@ public KafkaFailureHandler create(KafkaConnectorIncomingConfiguration config,
                             "lazy-client", c -> true,
                             CLIENT_ID_CONFIG, c -> "kafka-delayed-retry-topic-" + consumerClientId,
                             GROUP_ID_CONFIG, c -> "kafka-delayed-retry-topic-" + consumerClientId));
-            KafkaConnectorIncomingConfiguration retryConfig = new KafkaConnectorIncomingConfiguration(retryConsumerConfig);
+            Config retryKafkaConfig = ConfigHelper.retrieveChannelConfiguration(configurations, retryConsumerConfig);
+            KafkaConnectorIncomingConfiguration retryConfig = new KafkaConnectorIncomingConfiguration(retryKafkaConfig);
             ReactiveKafkaConsumer<Object, Object> retryConsumer = new ReactiveKafkaConsumer<>(retryConfig,
                     deserializationFailureHandlers,
                     retryConsumerConfig.getValue(GROUP_ID_CONFIG, String.class), -1,

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyFactory.java

@@ -2,6 +2,7 @@
 
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.util.Map;
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.enterprise.inject.Any;
@@ -70,9 +71,13 @@ public class KafkaRequestReplyFactory implements EmitterFactory<KafkaRequestRepl
     @Inject
     Instance<Config> config;
 
+    @Inject
+    @Any
+    Instance<Map<String, Object>> configurations;
+
     @Override
     public KafkaRequestReplyImpl<Object, Object> createEmitter(EmitterConfiguration configuration, long defaultBufferSize) {
-        return new KafkaRequestReplyImpl<>(configuration, defaultBufferSize, config.get(), holder.vertx(),
+        return new KafkaRequestReplyImpl<>(configuration, defaultBufferSize, config.get(), configurations, holder.vertx(),
                 kafkaCDIEvents, commitStrategyFactories, failureStrategyFactories, failureHandlers,
                 correlationIdHandlers, replyFailureHandlers, rebalanceListeners);
     }

smallrye-reactive-messaging-kafka/src/main/java/io/smallrye/reactive/messaging/kafka/reply/KafkaRequestReplyImpl.java

@@ -43,6 +43,7 @@
 import io.smallrye.reactive.messaging.kafka.api.OutgoingKafkaRecordMetadata;
 import io.smallrye.reactive.messaging.kafka.commit.KafkaCommitHandler;
 import io.smallrye.reactive.messaging.kafka.fault.KafkaFailureHandler;
+import io.smallrye.reactive.messaging.kafka.impl.ConfigHelper;
 import io.smallrye.reactive.messaging.kafka.impl.KafkaSource;
 import io.smallrye.reactive.messaging.kafka.impl.TopicPartitions;
 import io.smallrye.reactive.messaging.providers.extension.MutinyEmitterImpl;
@@ -75,7 +76,8 @@ public class KafkaRequestReplyImpl<Req, Rep> extends MutinyEmitterImpl<Req>
 
     public KafkaRequestReplyImpl(EmitterConfiguration config,
             long defaultBufferSize,
-            Config rootConfig,
+            Config channelConfiguration,
+            Instance<Map<String, Object>> configurations,
             Vertx vertx,
             KafkaCDIEvents kafkaCDIEvents,
             Instance<KafkaCommitHandler.Factory> commitHandlerFactory,
@@ -86,12 +88,13 @@ public KafkaRequestReplyImpl(EmitterConfiguration config,
             Instance<KafkaConsumerRebalanceListener> rebalanceListeners) {
         super(config, defaultBufferSize);
         this.channel = config.name();
-        ConnectorConfig connectorConfig = new OverrideConnectorConfig(OUTGOING_PREFIX, rootConfig, channel,
+        ConnectorConfig connectorConfig = new OverrideConnectorConfig(OUTGOING_PREFIX, channelConfiguration, channel,
                 "reply", Map.of(
                         "topic", c -> c.getOriginalValue("topic", String.class).orElse(channel) + DEFAULT_REPLIES_TOPIC_SUFFIX,
                         "assign-seek",
                         c -> c.getOriginalValue(REPLY_PARTITION_KEY, Integer.class).map(String::valueOf).orElse(null)));
-        KafkaConnectorIncomingConfiguration consumerConfig = new KafkaConnectorIncomingConfiguration(connectorConfig);
+        Config replyKafkaConfig = ConfigHelper.retrieveChannelConfiguration(configurations, connectorConfig);
+        KafkaConnectorIncomingConfiguration consumerConfig = new KafkaConnectorIncomingConfiguration(replyKafkaConfig);
         this.replyTopic = consumerConfig.getTopic().orElse(null);
         this.replyPartition = connectorConfig.getOptionalValue(REPLY_PARTITION_KEY, Integer.class).orElse(-1);
         this.replyTimeout = Duration.ofMillis(connectorConfig.getOptionalValue(REPLY_TIMEOUT_KEY, Integer.class).orElse(5000));