Repository: quarkusio/quarkus
Issue: Config like `mp.messaging.connector.smallrye-kafka.value.serializer` is ignored

### Describe the bug

The build step `SmallRyeReactiveMessagingKafkaProcessor#defaultChannelConfiguration()` currently ignores checking for connector-level config like `mp.messaging.connector.smallrye-kafka.value.serializer`. As a result this configuration has no effect when added to the `application.properties` file because the generated channel-level `RunTimeConfigurationDefaultBuildItem` defaults take precedence.

See also discussion in Zulip: https://quarkusio.zulipchat.com/#narrow/stream/187030-users/topic/.E2.9C.94.20Kafka.3A.20Apicurio.20for.20dev.20and.20Confluent.20for.20prod.20mode.3F

### Expected behavior

Explicit connector-level config in `application.properties` should take precedence over generated channel-level defaults.

The build step should probably check for the presence of such properties at build time to determine whether to generate a runtime default or not for a given channel's properties.

#### Use case

We are using Kafka with a schema registry and Avro encoded messages. In production the schema registry is the Confluent implementation and for development in dev mode we use Apicurio. This means that the serializer and deserializer implementation classes are different for production and non-production (i.e. test and development) launch modes. Further, since our application uses multiple channels, we would like to declare the Avro serializer and deserializer at the connector level, rather than individually for each channel. Doing it for each channel results in a lot of duplicated config.

### Actual behavior

The explicit connector-level config (like `mp.messaging.connector.smallrye-kafka.value.serializer`) in the `application.properties` file is ignored, because the autodetection generates corresponding channel-level config which takes precedence.

### How to Reproduce?

_No response_

### Output of `uname -a` or `ver`

_No response_

### Output of `java -version`

_No response_

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

2.7.4.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

_No response_

### Additional information


Changed files:
extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java
extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java
extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java

diff --git a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java
index cda62e4596c..555a611e175 100644
--- a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java
+++ b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeDiscoveryState.java
@@ -8,6 +8,7 @@
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import org.eclipse.microprofile.config.Config;
 import org.eclipse.microprofile.config.ConfigProvider;
 import org.jboss.jandex.AnnotationInstance;
 import org.jboss.jandex.AnnotationTarget;
@@ -26,6 +27,11 @@ class DefaultSerdeDiscoveryState {
     private final Map<String, Boolean> isKafkaConnector = new HashMap<>();
     private final Set<String> alreadyConfigured = new HashSet<>();
 
+    private Boolean connectorHasKeySerializer;
+    private Boolean connectorHasValueSerializer;
+    private Boolean connectorHasKeyDeserializer;
+    private Boolean connectorHasValueDeserializer;
+
     private Boolean hasConfluent;
     private Boolean hasApicurio1;
     private Boolean hasApicurio2;
@@ -35,6 +41,10 @@ class DefaultSerdeDiscoveryState {
         this.index = index;
     }
 
+    Config getConfig() {
+        return ConfigProvider.getConfig();
+    }
+
     boolean isKafkaConnector(List<ConnectorManagedChannelBuildItem> channelsManagedByConnectors, boolean incoming,
             String channelName) {
         // First look in the channelsManagedByConnectors list
@@ -48,13 +58,61 @@ boolean isKafkaConnector(List<ConnectorManagedChannelBuildItem> channelsManagedB
         String channelType = incoming ? "incoming" : "outgoing";
         return isKafkaConnector.computeIfAbsent(channelType + "|" + channelName, ignored -> {
             String connectorKey = "mp.messaging." + channelType + "." + channelName + ".connector";
-            String connector = ConfigProvider.getConfig()
+            String connector = getConfig()
                     .getOptionalValue(connectorKey, String.class)
                     .orElse("ignored");
             return KafkaConnector.CONNECTOR_NAME.equals(connector);
         });
     }
 
+    boolean shouldNotConfigure(String key) {
+        // if we know at build time that key/value [de]serializer is configured on the connector,
+        // we should NOT emit default configuration for key/value [de]serializer on the channel
+        // (in other words, only a user can explicitly override a connector configuration)
+        //
+        // more config properties could possibly be handled in the same way, but these should suffice for now
+
+        if (key.startsWith("mp.messaging.outgoing.") && key.endsWith(".key.serializer")) {
+            if (connectorHasKeySerializer == null) {
+                connectorHasKeySerializer = getConfig()
+                        .getOptionalValue("mp.messaging.connector." + KafkaConnector.CONNECTOR_NAME + ".key.serializer",
+                                String.class)
+                        .isPresent();
+            }
+            return connectorHasKeySerializer;
+        }
+        if (key.startsWith("mp.messaging.outgoing.") && key.endsWith(".value.serializer")) {
+            if (connectorHasValueSerializer == null) {
+                connectorHasValueSerializer = getConfig()
+                        .getOptionalValue("mp.messaging.connector." + KafkaConnector.CONNECTOR_NAME + ".value.serializer",
+                                String.class)
+                        .isPresent();
+            }
+            return connectorHasValueSerializer;
+        }
+
+        if (key.startsWith("mp.messaging.incoming.") && key.endsWith(".key.deserializer")) {
+            if (connectorHasKeyDeserializer == null) {
+                connectorHasKeyDeserializer = getConfig()
+                        .getOptionalValue("mp.messaging.connector." + KafkaConnector.CONNECTOR_NAME + ".key.deserializer",
+                                String.class)
+                        .isPresent();
+            }
+            return connectorHasKeyDeserializer;
+        }
+        if (key.startsWith("mp.messaging.incoming.") && key.endsWith(".value.deserializer")) {
+            if (connectorHasValueDeserializer == null) {
+                connectorHasValueDeserializer = getConfig()
+                        .getOptionalValue("mp.messaging.connector." + KafkaConnector.CONNECTOR_NAME + ".value.deserializer",
+                                String.class)
+                        .isPresent();
+            }
+            return connectorHasValueDeserializer;
+        }
+
+        return false;
+    }
+
     void ifNotYetConfigured(String key, Runnable runnable) {
         if (!alreadyConfigured.contains(key)) {
             alreadyConfigured.add(key);
diff --git a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java
index 18513a1fb02..2770b0dc637 100644
--- a/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java
+++ b/extensions/smallrye-reactive-messaging-kafka/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/SmallRyeReactiveMessagingKafkaProcessor.java
@@ -243,6 +243,10 @@ private void produceRuntimeConfigurationDefaultBuildItem(DefaultSerdeDiscoverySt
             return;
         }
 
+        if (discovery.shouldNotConfigure(key)) {
+            return;
+        }
+
         discovery.ifNotYetConfigured(key, () -> {
             config.produce(new RunTimeConfigurationDefaultBuildItem(key, value));
         });
diff --git a/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java b/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java
index 328a837b928..ba94e489267 100644
--- a/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java
+++ b/extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/DefaultSerdeConfigTest.java
@@ -8,6 +8,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletionStage;
 
 import javax.inject.Inject;
@@ -19,6 +20,8 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.header.Headers;
 import org.assertj.core.groups.Tuple;
+import org.eclipse.microprofile.config.Config;
+import org.eclipse.microprofile.config.spi.ConfigProviderResolver;
 import org.eclipse.microprofile.reactive.messaging.Channel;
 import org.eclipse.microprofile.reactive.messaging.Emitter;
 import org.eclipse.microprofile.reactive.messaging.Incoming;
@@ -38,6 +41,8 @@
 import io.quarkus.kafka.client.serialization.JsonbSerializer;
 import io.quarkus.kafka.client.serialization.ObjectMapperDeserializer;
 import io.quarkus.smallrye.reactivemessaging.deployment.items.ConnectorManagedChannelBuildItem;
+import io.smallrye.config.SmallRyeConfigBuilder;
+import io.smallrye.config.common.MapBackedConfigSource;
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.Uni;
 import io.smallrye.reactive.messaging.MutinyEmitter;
@@ -48,20 +53,36 @@
 
 public class DefaultSerdeConfigTest {
     private static void doTest(Tuple[] expectations, Class<?>... classesToIndex) {
+        doTest(null, expectations, classesToIndex);
+    }
+
+    private static void doTest(Config customConfig, Tuple[] expectations, Class<?>... classesToIndex) {
         List<RunTimeConfigurationDefaultBuildItem> configs = new ArrayList<>();
 
         DefaultSerdeDiscoveryState discovery = new DefaultSerdeDiscoveryState(index(classesToIndex)) {
+            @Override
+            Config getConfig() {
+                return customConfig != null ? customConfig : super.getConfig();
+            }
+
             @Override
             boolean isKafkaConnector(List<ConnectorManagedChannelBuildItem> list, boolean incoming, String channelName) {
                 return true;
             }
         };
-        new SmallRyeReactiveMessagingKafkaProcessor().discoverDefaultSerdeConfig(discovery, Collections.emptyList(),
-                configs::add, null, null);
-
-        assertThat(configs)
-                .extracting(RunTimeConfigurationDefaultBuildItem::getKey, RunTimeConfigurationDefaultBuildItem::getValue)
-                .containsExactlyInAnyOrder(expectations);
+        try {
+            new SmallRyeReactiveMessagingKafkaProcessor().discoverDefaultSerdeConfig(discovery, Collections.emptyList(),
+                    configs::add, null, null);
+
+            assertThat(configs)
+                    .extracting(RunTimeConfigurationDefaultBuildItem::getKey, RunTimeConfigurationDefaultBuildItem::getValue)
+                    .containsExactlyInAnyOrder(expectations);
+        } finally {
+            // must not leak the Config instance associated to the system classloader
+            if (customConfig == null) {
+                ConfigProviderResolver.instance().releaseConfig(discovery.getConfig());
+            }
+        }
     }
 
     private static IndexView index(Class<?>... classes) {
@@ -2467,4 +2488,69 @@ void channel10(List<Message<JacksonDto>> jacksonDto) {
 
     }
 
+    // ---
+
+    @Test
+    public void connectorConfigNotOverriden() {
+        // @formatter:off
+        Tuple[] expectations1 = {
+                // "mp.messaging.outgoing.channel1.key.serializer" NOT expected, connector config exists
+                tuple("mp.messaging.outgoing.channel1.value.serializer",   "org.apache.kafka.common.serialization.StringSerializer"),
+
+                tuple("mp.messaging.incoming.channel2.key.deserializer",   "org.apache.kafka.common.serialization.LongDeserializer"),
+                // "mp.messaging.incoming.channel2.value.deserializer" NOT expected, connector config exists
+
+                tuple("mp.messaging.incoming.channel3.key.deserializer",   "org.apache.kafka.common.serialization.LongDeserializer"),
+                // "mp.messaging.incoming.channel3.value.deserializer" NOT expected, connector config exists
+                // "mp.messaging.outgoing.channel4.key.serializer" NOT expected, connector config exists
+                tuple("mp.messaging.outgoing.channel4.value.serializer",   "org.apache.kafka.common.serialization.StringSerializer"),
+        };
+
+        Tuple[] expectations2 = {
+                tuple("mp.messaging.outgoing.channel1.key.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
+                // "mp.messaging.outgoing.channel1.value.serializer" NOT expected, connector config exists
+
+                // "mp.messaging.incoming.channel2.key.deserializer" NOT expected, connector config exists
+                tuple("mp.messaging.incoming.channel2.value.deserializer",   "org.apache.kafka.common.serialization.StringDeserializer"),
+
+                // "mp.messaging.incoming.channel3.key.deserializer" NOT expected, connector config exists
+                tuple("mp.messaging.incoming.channel3.value.deserializer",   "org.apache.kafka.common.serialization.StringDeserializer"),
+                tuple("mp.messaging.outgoing.channel4.key.serializer",   "org.apache.kafka.common.serialization.LongSerializer"),
+                // "mp.messaging.outgoing.channel4.value.serializer" NOT expected, connector config exists
+        };
+        // @formatter:on
+
+        doTest(new SmallRyeConfigBuilder()
+                .withSources(new MapBackedConfigSource("test", Map.of(
+                        "mp.messaging.connector.smallrye-kafka.key.serializer", "foo.Bar",
+                        "mp.messaging.connector.smallrye-kafka.value.deserializer", "foo.Baz")) {
+                })
+                .build(), expectations1, ConnectorConfigNotOverriden.class);
+
+        doTest(new SmallRyeConfigBuilder()
+                .withSources(new MapBackedConfigSource("test", Map.of(
+                        "mp.messaging.connector.smallrye-kafka.key.deserializer", "foo.Bar",
+                        "mp.messaging.connector.smallrye-kafka.value.serializer", "foo.Baz")) {
+                })
+                .build(), expectations2, ConnectorConfigNotOverriden.class);
+    }
+
+    private static class ConnectorConfigNotOverriden {
+        @Outgoing("channel1")
+        Record<Long, String> method1() {
+            return null;
+        }
+
+        @Incoming("channel2")
+        CompletionStage<?> method2(Record<Long, String> msg) {
+            return null;
+        }
+
+        @Incoming("channel3")
+        @Outgoing("channel4")
+        Record<Long, String> method3(Record<Long, String> msg) {
+            return null;
+        }
+    }
+
 }