Repository: quarkusio/quarkus
Issue: Update to 2.0.1.Final breaks working Kafka deserializer configuration in 1.x

## Describe the bug
Changing pom.xml from 1.x to 2.0.1.Final prevents successful Dev Mode startup with exception.

### Expected behavior
No errors during startup like before with 1.x

### Actual behavior
Exception during startup is thrown:

```
2021-07-07 17:06:06,364 ERROR [io.qua.dep.dev.IsolatedDevModeMain] (main) Failed to start quarkus: java.lang.RuntimeException: io.quarkus.builder.BuildException: Build failure: Build failed due to errors
	[error]: Build step io.quarkus.deployment.steps.ConfigGenerationBuildStep#generateConfigClass threw an exception: java.lang.IllegalStateException: More than one default value for mp.messaging.incoming.created.value.deserializer was produced
	at io.quarkus.deployment.steps.ConfigGenerationBuildStep.generateConfigClass(ConfigGenerationBuildStep.java:64)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at io.quarkus.deployment.ExtensionLoader$2.execute(ExtensionLoader.java:820)
	at io.quarkus.builder.BuildContext.run(BuildContext.java:277)
	at org.jboss.threads.ContextHandler$1.runWith(ContextHandler.java:18)
	at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2442)
	at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1476)
	at java.base/java.lang.Thread.run(Thread.java:829)
	at org.jboss.threads.JBossThread.run(JBossThread.java:501)

	at io.quarkus.runner.bootstrap.AugmentActionImpl.runAugment(AugmentActionImpl.java:414)
	at io.quarkus.runner.bootstrap.AugmentActionImpl.createInitialRuntimeApplication(AugmentActionImpl.java:274)
	at io.quarkus.runner.bootstrap.AugmentActionImpl.createInitialRuntimeApplication(AugmentActionImpl.java:66)
	at io.quarkus.deployment.dev.IsolatedDevModeMain.firstStart(IsolatedDevModeMain.java:87)
	at io.quarkus.deployment.dev.IsolatedDevModeMain.accept(IsolatedDevModeMain.java:448)
	at io.quarkus.deployment.dev.IsolatedDevModeMain.accept(IsolatedDevModeMain.java:61)
	at io.quarkus.bootstrap.app.CuratedApplication.runInCl(CuratedApplication.java:132)
	at io.quarkus.bootstrap.app.CuratedApplication.runInAugmentClassLoader(CuratedApplication.java:89)
	at io.quarkus.deployment.dev.DevModeMain.start(DevModeMain.java:145)
	at io.quarkus.deployment.dev.DevModeMain.main(DevModeMain.java:63)
Caused by: io.quarkus.builder.BuildException: Build failure: Build failed due to errors
	[error]: Build step io.quarkus.deployment.steps.ConfigGenerationBuildStep#generateConfigClass threw an exception: java.lang.IllegalStateException: More than one default value for mp.messaging.incoming.created.value.deserializer was produced
	at io.quarkus.deployment.steps.ConfigGenerationBuildStep.generateConfigClass(ConfigGenerationBuildStep.java:64)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at io.quarkus.deployment.ExtensionLoader$2.execute(ExtensionLoader.java:820)
	at io.quarkus.builder.BuildContext.run(BuildContext.java:277)
	at org.jboss.threads.ContextHandler$1.runWith(ContextHandler.java:18)
	at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2442)
	at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1476)
	at java.base/java.lang.Thread.run(Thread.java:829)
	at org.jboss.threads.JBossThread.run(JBossThread.java:501)

	at io.quarkus.builder.Execution.run(Execution.java:116)
	at io.quarkus.builder.BuildExecutionBuilder.execute(BuildExecutionBuilder.java:79)
	at io.quarkus.deployment.QuarkusAugmentor.run(QuarkusAugmentor.java:156)
	at io.quarkus.runner.bootstrap.AugmentActionImpl.runAugment(AugmentActionImpl.java:412)
	... 9 more
Caused by: java.lang.IllegalStateException: More than one default value for mp.messaging.incoming.created.value.deserializer was produced
	at io.quarkus.deployment.steps.ConfigGenerationBuildStep.generateConfigClass(ConfigGenerationBuildStep.java:64)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at io.quarkus.deployment.ExtensionLoader$2.execute(ExtensionLoader.java:820)
	at io.quarkus.builder.BuildContext.run(BuildContext.java:277)
	at org.jboss.threads.ContextHandler$1.runWith(ContextHandler.java:18)
	at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2442)
	at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1476)
	at java.base/java.lang.Thread.run(Thread.java:829)
	at org.jboss.threads.JBossThread.run(JBossThread.java:501)
```

## To Reproduce

Could not reproduce with quarkus-quickstarts/kafka-panache-reactive-quickstart, but there isn't any quickstart utilising deserializers.

### Configuration
according to [Serializing via JSON-B](https://quarkus.io/guides/kafka#jsonb-serialization)
```properties
# Add your application.properties here, if applicable.
# Configure the Kafka sink (we write to it)
mp.messaging.outgoing.entry-created.connector=smallrye-kafka
mp.messaging.outgoing.entry-created.topic=created
mp.messaging.outgoing.entry-created.value.serializer=io.quarkus.kafka.client.serialization.JsonbSerializer

# Configure the Kafka source (we read from it)
mp.messaging.incoming.created.connector=smallrye-kafka
mp.messaging.incoming.created.topic=created
mp.messaging.incoming.created.broadcast=true
mp.messaging.incoming.created.value.deserializer=me.cligg.dto.MyCacheEntryDeserializer
```

### Screenshots
(If applicable, add screenshots to help explain your problem.)

## Environment (please complete the following information):

### Output of `uname -a` or `ver`

Linux fedora 5.12.13-300.fc34.x86_64 #1 SMP Wed Jun 23 16:18:11 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux

### Output of `java -version`

openjdk version "11.0.11" 2021-04-20
OpenJDK Runtime Environment 18.9 (build 11.0.11+9)
OpenJDK 64-Bit Server VM 18.9 (build 11.0.11+9, mixed mode, sharing)

### GraalVM version (if different from Java)

### Quarkus version or git rev

    <quarkus-plugin.version>2.0.1.Final</quarkus-plugin.version>
    <quarkus.platform.version>2.0.1.Final</quarkus.platform.version>

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

Apache Maven 3.8.1 (05c21c65bdfed0f71a2f2ada8b84da59348c4c5d)
Maven home: /home/torben/.m2/wrapper/dists/apache-maven-3.8.1-bin/2l5mhf2pq2clrde7f7qp1rdt5m/apache-maven-3.8.1
Java version: 11.0.11, vendor: Red Hat, Inc., runtime: /usr/lib/jvm/java-11-openjdk-11.0.11.0.9-4.fc34.x86_64
Default locale: en_US, platform encoding: UTF-8
OS name: "linux", version: "5.12.13-300.fc34.x86_64", arch: "amd64", family: "unix"

## Additional context
(Add any other context about the problem here.)


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 f4740fe5924..b6bade7d88c 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
@@ -19,6 +19,7 @@ class DefaultSerdeDiscoveryState {
     private final IndexView index;
 
     private final Map<String, Boolean> isKafkaConnector = new HashMap<>();
+    private final Map<String, String> serdeConfigMap = new HashMap<>();
 
     private Boolean hasConfluent;
     private Boolean hasApicurio1;
@@ -40,6 +41,13 @@ boolean isKafkaConnector(boolean incoming, String channelName) {
         });
     }
 
+    void runIfConfigIsAbsent(String key, String value, Runnable runnable) {
+        if (value != null && !serdeConfigMap.containsKey(key)) {
+            serdeConfigMap.put(key, value);
+            runnable.run();
+        }
+    }
+
     boolean isAvroGenerated(DotName className) {
         ClassInfo clazz = index.getClassByName(className);
         return clazz != null && clazz.classAnnotation(DotNames.AVRO_GENERATED) != null;
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 2687f5f4a0f..3fbec6398fe 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
@@ -65,14 +65,10 @@ void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery,
             MethodInfo method = annotation.target().asMethod();
 
             processIncomingMethod(discovery, method, (keyDeserializer, valueDeserializer) -> {
-                if (keyDeserializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.incoming." + channelName + ".key.deserializer", keyDeserializer));
-                }
-                if (valueDeserializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.incoming." + channelName + ".value.deserializer", valueDeserializer));
-                }
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.incoming." + channelName + ".key.deserializer", keyDeserializer);
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.incoming." + channelName + ".value.deserializer", valueDeserializer);
             });
         }
 
@@ -85,14 +81,10 @@ void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery,
             MethodInfo method = annotation.target().asMethod();
 
             processOutgoingMethod(discovery, method, (keySerializer, valueSerializer) -> {
-                if (keySerializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.outgoing." + channelName + ".key.serializer", keySerializer));
-                }
-                if (valueSerializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.outgoing." + channelName + ".value.serializer", valueSerializer));
-                }
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.outgoing." + channelName + ".key.serializer", keySerializer);
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.outgoing." + channelName + ".value.serializer", valueSerializer);
             });
         }
 
@@ -117,29 +109,27 @@ void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery,
             }
 
             processIncomingChannelInjectionPoint(discovery, injectionPointType, (keyDeserializer, valueDeserializer) -> {
-                if (keyDeserializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.incoming." + channelName + ".key.deserializer", keyDeserializer));
-                }
-                if (valueDeserializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.incoming." + channelName + ".value.deserializer", valueDeserializer));
-                }
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.incoming." + channelName + ".key.deserializer", keyDeserializer);
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.incoming." + channelName + ".value.deserializer", valueDeserializer);
             });
 
             processOutgoingChannelInjectionPoint(discovery, injectionPointType, (keySerializer, valueSerializer) -> {
-                if (keySerializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.outgoing." + channelName + ".key.serializer", keySerializer));
-                }
-                if (valueSerializer != null) {
-                    config.produce(new RunTimeConfigurationDefaultBuildItem(
-                            "mp.messaging.outgoing." + channelName + ".value.serializer", valueSerializer));
-                }
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.outgoing." + channelName + ".key.serializer", keySerializer);
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config,
+                        "mp.messaging.outgoing." + channelName + ".value.serializer", valueSerializer);
             });
         }
     }
 
+    void produceRuntimeConfigurationDefaultBuildItem(DefaultSerdeDiscoveryState discovery,
+            BuildProducer<RunTimeConfigurationDefaultBuildItem> config, String key, String value) {
+        discovery.runIfConfigIsAbsent(key, value,
+                () -> config.produce(new RunTimeConfigurationDefaultBuildItem(key, value)));
+    }
+
     private void processIncomingMethod(DefaultSerdeDiscoveryState discovery, MethodInfo method,
             BiConsumer<String, String> deserializerAcceptor) {
         List<Type> parameterTypes = method.parameters();
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 b4c3dcfe3c3..69a35ba4c73 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
@@ -55,7 +55,7 @@ boolean isKafkaConnector(boolean incoming, String channelName) {
 
         assertThat(configs)
                 .extracting(RunTimeConfigurationDefaultBuildItem::getKey, RunTimeConfigurationDefaultBuildItem::getValue)
-                .containsOnly(expectations);
+                .containsExactlyInAnyOrder(expectations);
     }
 
     private static IndexView index(Class<?>... classes) {
@@ -1977,4 +1977,62 @@ private static class FloatJsonArrayInShortByteArrayOut {
         @Channel("channel25")
         Multi<ConsumerRecord<Float, JsonArray>> consumer25;
     }
+
+    @Test
+    void produceDefaultConfigOnce() {
+        // @formatter:off
+        Tuple[] expectations = {
+                tuple("mp.messaging.outgoing.channel1.value.serializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JsonbDtoSerializer"),
+                tuple("mp.messaging.incoming.channel2.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
+                tuple("mp.messaging.incoming.channel3.key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
+                tuple("mp.messaging.incoming.channel3.value.deserializer", "io.quarkus.smallrye.reactivemessaging.kafka.deployment.DefaultSerdeConfigTest$JacksonDtoDeserializer"),
+                tuple("mp.messaging.outgoing.channel4.key.serializer", "org.apache.kafka.common.serialization.StringSerializer"),
+                tuple("mp.messaging.outgoing.channel4.value.serializer", "org.apache.kafka.common.serialization.IntegerSerializer"),
+        };
+        // @formatter:on
+
+        doTest(expectations, JsonbDto.class, JsonbDtoSerializer.class, JacksonDto.class, JacksonDtoDeserializer.class,
+                MultipleChannels.class);
+    }
+
+    private static class MultipleChannels {
+
+        @Channel("channel1")
+        Emitter<JsonbDto> emitter1;
+
+        @Outgoing("channel1")
+        Publisher<Message<JsonbDto>> method1() {
+            return null;
+        }
+
+        @Outgoing("channel1")
+        Publisher<JsonbDto> method1Duplicate() {
+            return null;
+        }
+
+        @Channel("channel2")
+        Multi<JacksonDto> channel2;
+
+        @Incoming("channel2")
+        void channel2Duplicate(JacksonDto jacksonDto) {
+
+        }
+
+        @Channel("channel3")
+        Multi<Record<Integer, JacksonDto>> channel3;
+
+        @Incoming("channel3")
+        void channel3Duplicate(Record<Integer, JacksonDto> jacksonDto) {
+
+        }
+
+        @Channel("channel4")
+        Emitter<ProducerRecord<String, Integer>> emitterChannel4;
+
+        @Outgoing("channel4")
+        ProducerRecord<String, Integer> method4() {
+            return null;
+        };
+
+    }
 }