Repository: quarkusio/quarkus
Issue: Smallrye Reactive Messaging with Kafka Connector does not support channel names containing a dot

### Describe the bug

Seems like that a dot in the channel name isn't supported when using ``Smallrye Reactive Messaging`` and ``Kafka``:

```
@ApplicationScoped
public class TestConsumer {

  @Incoming("test.test")
  public void consume(ChangeEvent event) {
    Log.info(event);
  }
}
```

This leads to the following exception: 
``(Quarkus Main Thread) Failed to start application (with profile [dev]): java.lang.IllegalArgumentException: The attribute `value.deserializer` on connector 'smallrye-kafka' (channel: test.test) must be set``

### Expected behavior

Quarkus should behave in the same way as defining a channel name without a dot like ``test``:

```
2023-05-31 14:50:31,122 INFO  [io.qua.sma.dep.processor] (build-22) Configuring the channel 'test' to be managed by the connector 'smallrye-kafka'
2023-05-31 14:50:31,125 INFO  [io.qua.sma.dep.processor] (build-37) Generating Jackson deserializer for type com........event.consumer.ChangeEvent
2023-05-31 14:50:35,103 INFO  [io.qua.kaf.cli.dep.DevServicesKafkaProcessor] (build-2) Dev Services for Kafka started. Other Quarkus applications in dev mode will find the broker automatically. For Quarkus applications in production mode, you can connect to this by starting your application with -Dkafka.bootstr
ap.servers=OUTSIDE://localhost:32777
__  ____  __  _____   ___  __ ____  ______
 --/ __ \\/ / / / _ | / _ \\/ //_/ / / / __/
 -/ /_/ / /_/ / __ |/ , _/ ,< / /_/ /\\ \\
--\\___\\_\\____/_/ |_/_/|_/_/|_|\\____/___/


2023-05-31 14:50:35,854 INFO  [io.sma.rea.mes.kafka] (Quarkus Main Thread) SRMSG18229: Configured topics for channel 'test': [test]
2023-05-31 14:50:35,859 INFO  [io.sma.rea.mes.kafka] (Quarkus Main Thread) SRMSG18214: Key deserializer omitted, using String as default
2023-05-31 14:50:35,993 INFO  [io.ope.api.GlobalOpenTelemetry] (Quarkus Main Thread) AutoConfiguredOpenTelemetrySdk found on classpath but automatic configuration is disabled. To enable, run your JVM with -Dotel.java.global-autoconfigure.enabled=true
2023-05-31 14:50:36,011 INFO  [io.sma.rea.mes.kafka] (smallrye-kafka-consumer-thread-0) SRMSG18257: Kafka consumer kafka-consumer-test, connected to Kafka brokers 'OUTSIDE://localhost:32777', belongs to the 'eventbatching' consumer group and is configured to poll records from [test]
2023-05-31 14:50:36,071 INFO  [io.quarkus] (Quarkus Main Thread) eventbatching main-SNAPSHOT on JVM (powered by Quarkus 3.0.4.Final) started in 5.846s. Listening on: http://localhost:8080
2023-05-31 14:50:36,072 INFO  [io.quarkus] (Quarkus Main Thread) Profile dev activated. Live Coding activated.
2023-05-31 14:50:36,072 INFO  [io.quarkus] (Quarkus Main Thread) Installed features: [cdi, jdbc-postgresql, kafka-client, logging-json, micrometer, opentelemetry, resteasy-reactive, resteasy-reactive-jackson, smallrye-context-propagation, smallrye-health, smallrye-reactive-messaging, smallrye-reactive-messaging
-kafka, vertx]

```


### Actual behavior

The startup fails due to the following exception:

```
2023-05-31 14:44:41,302 ERROR [io.qua.run.Application] (Quarkus Main Thread) Failed to start application (with profile [dev]): java.lang.IllegalArgumentException: The attribute `value.deserializer` on connector 'smallrye-kafka' (channel: test.test) must be set
        at io.smallrye.reactive.messaging.kafka.KafkaConnectorIncomingConfiguration.lambda$getValueDeserializer$0(KafkaConnectorIncomingConfiguration.java:62)
        at java.base/java.util.Optional.orElseThrow(Optional.java:403)
        at io.smallrye.reactive.messaging.kafka.KafkaConnectorIncomingConfiguration.getValueDeserializer(KafkaConnectorIncomingConfiguration.java:62)
        at io.smallrye.reactive.messaging.kafka.KafkaConnectorIncomingConfiguration.validate(KafkaConnectorIncomingConfiguration.java:425)
        at io.smallrye.reactive.messaging.kafka.KafkaConnectorIncomingConfiguration.<init>(KafkaConnectorIncomingConfiguration.java:16)
        at io.smallrye.reactive.messaging.kafka.KafkaConnector.getPublisher(KafkaConnector.java:189)
        at io.smallrye.reactive.messaging.kafka.KafkaConnector_Subclass.getPublisher$$superforward(Unknown Source)
        at io.smallrye.reactive.messaging.kafka.KafkaConnector_Subclass$$function$$9.apply(Unknown Source)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:74)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext$NextAroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:100)
        at io.quarkus.smallrye.reactivemessaging.runtime.devmode.DevModeSupportConnectorFactoryInterceptor.intercept(DevModeSupportConnectorFactoryInterceptor.java:53)
        at io.quarkus.smallrye.reactivemessaging.runtime.devmode.DevModeSupportConnectorFactoryInterceptor_Bean.intercept(Unknown Source)
        at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:42)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:71)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.proceed(AroundInvokeInvocationContext.java:63)
        at io.quarkus.smallrye.reactivemessaging.runtime.DuplicatedContextConnectorFactoryInterceptor.intercept(DuplicatedContextConnectorFactoryInterceptor.java:32)
        at io.quarkus.smallrye.reactivemessaging.runtime.DuplicatedContextConnectorFactoryInterceptor_Bean.intercept(Unknown Source)
        at io.quarkus.arc.impl.InterceptorInvocation.invoke(InterceptorInvocation.java:42)
        at io.quarkus.arc.impl.AroundInvokeInvocationContext.perform(AroundInvokeInvocationContext.java:38)
        at io.quarkus.arc.impl.InvocationContexts.performAroundInvoke(InvocationContexts.java:26)
        at io.smallrye.reactive.messaging.kafka.KafkaConnector_Subclass.getPublisher(Unknown Source)
        at io.smallrye.reactive.messaging.kafka.KafkaConnector_ClientProxy.getPublisher(Unknown Source)
        at io.smallrye.reactive.messaging.providers.impl.ConfiguredChannelFactory.createPublisher(ConfiguredChannelFactory.java:172)
        at io.smallrye.reactive.messaging.providers.impl.ConfiguredChannelFactory.register(ConfiguredChannelFactory.java:134)
        at io.smallrye.reactive.messaging.providers.impl.ConfiguredChannelFactory.initialize(ConfiguredChannelFactory.java:106)
        at io.smallrye.reactive.messaging.providers.impl.ConfiguredChannelFactory_ClientProxy.initialize(Unknown Source)
        at java.base/java.util.Iterator.forEachRemaining(Iterator.java:133)
        at java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1845)
        at java.base/java.util.stream.ReferencePipeline$Head.forEach(ReferencePipeline.java:762)
        at io.smallrye.reactive.messaging.providers.extension.MediatorManager.start(MediatorManager.java:206)
        at io.smallrye.reactive.messaging.providers.extension.MediatorManager_ClientProxy.start(Unknown Source)
        at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle.onApplicationStart(SmallRyeReactiveMessagingLifecycle.java:52)
        at io.quarkus.smallrye.reactivemessaging.runtime.SmallRyeReactiveMessagingLifecycle_Observer_onApplicationStart_68e7b57eb97cb75d597c5b816682366e888d0d9b.notify(Unknown Source)
        at io.quarkus.arc.impl.EventImpl$Notifier.notifyObservers(EventImpl.java:346)
        at io.quarkus.arc.impl.EventImpl$Notifier.notify(EventImpl.java:328)
        at io.quarkus.arc.impl.EventImpl.fire(EventImpl.java:82)
        at io.quarkus.arc.runtime.ArcRecorder.fireLifecycleEvent(ArcRecorder.java:155)
        at io.quarkus.arc.runtime.ArcRecorder.handleLifecycleEvents(ArcRecorder.java:106)
        at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent1144526294.deploy_0(Unknown Source)
        at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent1144526294.deploy(Unknown Source)
        at io.quarkus.runner.ApplicationImpl.doStart(Unknown Source)
        at io.quarkus.runtime.Application.start(Application.java:101)
        at io.quarkus.runtime.ApplicationLifecycleManager.run(ApplicationLifecycleManager.java:111)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:71)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:44)
        at io.quarkus.runtime.Quarkus.run(Quarkus.java:124)
        at io.quarkus.runner.GeneratedMain.main(Unknown Source)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
        at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77)
        at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
        at java.base/java.lang.reflect.Method.invoke(Method.java:568)
        at io.quarkus.runner.bootstrap.StartupActionImpl$1.run(StartupActionImpl.java:104)
        at java.base/java.lang.Thread.run(Thread.java:833)
```

### How to Reproduce?

Try to run the following code snippet:

```
@ApplicationScoped
public class TestConsumer {

  @Incoming("test.test")
  public void consume(ChangeEvent event) {
    Log.info(event);
  }
}
```


### Output of `uname -a` or `ver`

_No response_

### Output of `java -version`

17

### GraalVM version (if different from Java)

_No response_

### Quarkus version or git rev

3.0.4.Final

### Build tool (ie. output of `mvnw --version` or `gradlew --version`)

mvn 3.8.8

### Additional information

_No response_

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 7e49c499fba..dbe9ede291e 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
@@ -1,5 +1,7 @@
 package io.quarkus.smallrye.reactivemessaging.kafka.deployment;
 
+import static io.quarkus.smallrye.reactivemessaging.kafka.deployment.SmallRyeReactiveMessagingKafkaProcessor.getChannelPropertyKey;
+
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -57,7 +59,7 @@ boolean isKafkaConnector(List<ConnectorManagedChannelBuildItem> channelsManagedB
 
         String channelType = incoming ? "incoming" : "outgoing";
         return isKafkaConnector.computeIfAbsent(channelType + "|" + channelName, ignored -> {
-            String connectorKey = "mp.messaging." + channelType + "." + channelName + ".connector";
+            String connectorKey = getChannelPropertyKey(channelName, "connector", incoming);
             String connector = getConfig()
                     .getOptionalValue(connectorKey, String.class)
                     .orElse("ignored");
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 ae4319516a4..d974580b1a6 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
@@ -102,6 +102,13 @@ private static List<String> getChannelProperties(String keySuffix, Config config
         return values;
     }
 
+    static String channelPropertyFormat = "mp.messaging.%s.%s.%s";
+
+    static String getChannelPropertyKey(String channelName, String propertyName, boolean incoming) {
+        return String.format(channelPropertyFormat, incoming ? "incoming" : "outgoing",
+                channelName.contains(".") ? "\\"" + channelName + "\\"" : channelName, propertyName);
+    }
+
     @BuildStep
     public void checkpointRedis(BuildProducer<AdditionalBeanBuildItem> additionalBean,
             BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
@@ -173,7 +180,7 @@ public void defaultChannelConfiguration(
                     if (!discoveryState.isKafkaConnector(channelsManagedByConnectors, true, channelName)) {
                         continue;
                     }
-                    String key = "mp.messaging.incoming." + channelName + ".graceful-shutdown";
+                    String key = getChannelPropertyKey(channelName, "graceful-shutdown", true);
                     discoveryState.ifNotYetConfigured(key, () -> {
                         defaultConfigProducer.produce(new RunTimeConfigurationDefaultBuildItem(key, "false"));
                     });
@@ -215,12 +222,11 @@ void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery,
             Type outgoingType = getOutgoingTypeFromMethod(method);
             processOutgoingType(discovery, outgoingType, (keySerializer, valueSerializer) -> {
                 produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                        "mp.messaging.outgoing." + channelName + ".key.serializer", keySerializer);
+                        getChannelPropertyKey(channelName, "key.serializer", false), keySerializer);
                 produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                        "mp.messaging.outgoing." + channelName + ".value.serializer", valueSerializer);
+                        getChannelPropertyKey(channelName, "value.serializer", false), valueSerializer);
 
-                handleAdditionalProperties("mp.messaging.outgoing." + channelName + ".", discovery,
-                        config, keySerializer, valueSerializer);
+                handleAdditionalProperties(channelName, false, discovery, config, keySerializer, valueSerializer);
             }, generatedClass, reflection, alreadyGeneratedSerializers);
         }
 
@@ -246,12 +252,11 @@ void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery,
             Type outgoingType = getOutgoingTypeFromChannelInjectionPoint(injectionPointType);
             processOutgoingType(discovery, outgoingType, (keySerializer, valueSerializer) -> {
                 produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                        "mp.messaging.outgoing." + channelName + ".key.serializer", keySerializer);
+                        getChannelPropertyKey(channelName, "key.serializer", false), keySerializer);
                 produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                        "mp.messaging.outgoing." + channelName + ".value.serializer", valueSerializer);
+                        getChannelPropertyKey(channelName, "value.serializer", false), valueSerializer);
 
-                handleAdditionalProperties("mp.messaging.outgoing." + channelName + ".", discovery,
-                        config, keySerializer, valueSerializer);
+                handleAdditionalProperties(channelName, false, discovery, config, keySerializer, valueSerializer);
             }, generatedClass, reflection, alreadyGeneratedSerializers);
         }
     }
@@ -259,17 +264,17 @@ void discoverDefaultSerdeConfig(DefaultSerdeDiscoveryState discovery,
     private void processKafkaTransactions(DefaultSerdeDiscoveryState discovery,
             BuildProducer<RunTimeConfigurationDefaultBuildItem> config, String channelName, Type injectionPointType) {
         if (injectionPointType != null && isKafkaEmitter(injectionPointType)) {
+            String transactionalIdKey = getChannelPropertyKey(channelName, "transactional.id", false);
+            String enableIdempotenceKey = getChannelPropertyKey(channelName, "enable.idempotence", false);
+            String acksKey = getChannelPropertyKey(channelName, "acks", false);
             LOGGER.infof("Transactional producer detected for channel '%s', setting following default config values: "
-                    + "'mp.messaging.outgoing.%s.transactional.id=${quarkus.application.name}-${channelName}', "
-                    + "'mp.messaging.outgoing.%s.enable.idempotence=true', "
-                    + "'mp.messaging.outgoing.%s.acks=all'", channelName, channelName, channelName, channelName);
-            produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                    "mp.messaging.outgoing." + channelName + ".transactional.id",
+                    + "'" + transactionalIdKey + "=${quarkus.application.name}-${channelName}', "
+                    + "'" + enableIdempotenceKey + "=true', "
+                    + "'" + acksKey + "=all'", channelName);
+            produceRuntimeConfigurationDefaultBuildItem(discovery, config, transactionalIdKey,
                     "${quarkus.application.name}-" + channelName);
-            produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                    "mp.messaging.outgoing." + channelName + ".enable.idempotence", "true");
-            produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                    "mp.messaging.outgoing." + channelName + ".acks", "all");
+            produceRuntimeConfigurationDefaultBuildItem(discovery, config, enableIdempotenceKey, "true");
+            produceRuntimeConfigurationDefaultBuildItem(discovery, config, acksKey, "all");
         }
     }
 
@@ -283,16 +288,15 @@ private void processIncomingType(DefaultSerdeDiscoveryState discovery,
                     alreadyGeneratedDeserializers);
 
             produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                    "mp.messaging.incoming." + channelName + ".key.deserializer", keyDeserializer);
+                    getChannelPropertyKey(channelName, "key.deserializer", true), keyDeserializer);
             produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                    "mp.messaging.incoming." + channelName + ".value.deserializer", valueDeserializer);
+                    getChannelPropertyKey(channelName, "value.deserializer", true), valueDeserializer);
             if (Boolean.TRUE.equals(isBatchType)) {
                 produceRuntimeConfigurationDefaultBuildItem(discovery, config,
-                        "mp.messaging.incoming." + channelName + ".batch", "true");
+                        getChannelPropertyKey(channelName, "batch", true), "true");
             }
 
-            handleAdditionalProperties("mp.messaging.incoming." + channelName + ".", discovery,
-                    config, keyDeserializer, valueDeserializer);
+            handleAdditionalProperties(channelName, true, discovery, config, keyDeserializer, valueDeserializer);
         });
     }
 
@@ -308,7 +312,7 @@ private Type getInjectionPointType(AnnotationInstance annotation) {
         }
     }
 
-    private void handleAdditionalProperties(String configPropertyBase, DefaultSerdeDiscoveryState discovery,
+    private void handleAdditionalProperties(String channelName, boolean incoming, DefaultSerdeDiscoveryState discovery,
             BuildProducer<RunTimeConfigurationDefaultBuildItem> config, Result... results) {
         for (Result result : results) {
             if (result == null) {
@@ -316,7 +320,8 @@ private void handleAdditionalProperties(String configPropertyBase, DefaultSerdeD
             }
 
             result.additionalProperties.forEach((key, value) -> {
-                produceRuntimeConfigurationDefaultBuildItem(discovery, config, configPropertyBase + key, value);
+                String configKey = getChannelPropertyKey(channelName, key, incoming);
+                produceRuntimeConfigurationDefaultBuildItem(discovery, config, configKey, value);
             });
         }
     }
@@ -946,7 +951,9 @@ private void processAnnotationsForReflectiveClassPayload(IndexView index, Config
     }
 
     private boolean isSerdeJson(IndexView index, Config config, String channelName, boolean serializer, boolean isKey) {
-        ConfigValue configValue = config.getConfigValue(getConfigName(channelName, serializer, isKey));
+        String configKey = getChannelPropertyKey(channelName, (isKey ? "key" : "value") + "." +
+                (serializer ? "serializer" : "deserializer"), !serializer);
+        ConfigValue configValue = config.getConfigValue(configKey);
         if (configValue.getValue() != null) {
             DotName serdeName = DotName.createSimple(configValue.getValue());
             return serializer ? isSubclassOfJsonSerializer(index, serdeName) : isSubclassOfJsonDeserializer(index, serdeName);
@@ -954,14 +961,6 @@ private boolean isSerdeJson(IndexView index, Config config, String channelName,
         return false;
     }
 
-    String getConfigName(String channelName, boolean serializer, boolean isKey) {
-        return "mp.messaging." +
-                (serializer ? "outgoing" : "incoming") + "." +
-                channelName + "." +
-                (isKey ? "key" : "value") + "." +
-                (serializer ? "serializer" : "deserializer");
-    }
-
     private boolean isSubclassOfJsonSerializer(IndexView index, DotName serializerName) {
         return isSubclassOf(index, DotNames.OBJECT_MAPPER_SERIALIZER, serializerName) ||
                 isSubclassOf(index, DotNames.JSONB_SERIALIZER, serializerName);
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 0f941cc60d1..70a6d866a70 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
@@ -2727,5 +2727,32 @@ void method2(JsonObject msg) {
 
     }
 
+    @Test
+    void channelNameContainingDot() {
+        Tuple[] expectations = {
+                tuple("mp.messaging.incoming.\\"new.channel\\".key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer"),
+                tuple("mp.messaging.incoming.\\"new.channel\\".value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"),
+                tuple("mp.messaging.outgoing.\\"new.channel.out\\".key.serializer", "org.apache.kafka.common.serialization.LongSerializer"),
+                tuple("mp.messaging.outgoing.\\"new.channel.out\\".value.serializer", "io.quarkus.kafka.client.serialization.JsonObjectSerializer"),
+                tuple("mp.messaging.outgoing.\\"new.channel.out\\".transactional.id", "${quarkus.application.name}-new.channel.out"),
+                tuple("mp.messaging.outgoing.\\"new.channel.out\\".enable.idempotence", "true"),
+                tuple("mp.messaging.outgoing.\\"new.channel.out\\".acks", "all"),
+        };
+        doTest(expectations, ChannelContainingDot.class);
+    }
+
+
+    private static class ChannelContainingDot {
+
+        @Incoming("new.channel")
+        void method1(KafkaRecord<Integer, String> msg) {
+
+        }
+
+        @Channel("new.channel.out")
+        KafkaTransactions<ProducerRecord<Long, JsonObject>> transactions;
+
+    }
+
 
 }