Messaging extensions doc tls-registry usage update

docs/src/main/asciidoc/amqp-reference.adoc

@@ -478,6 +478,19 @@ public AmqpClientOptions getNamedOptions() {
 }
 ----
 
+== TLS Configuration
+
+AMQP 1.0 Messaging extension integrates with the xref:./tls-registry-reference.adoc[Quarkus TLS registry] to configure the Vert.x AMQP client.
+
+To configure the TLS for an AMQP 1.0 channel, you need to provide a named TLS configuration in the `application.properties`:
+
+[source, properties]
+----
+quarkus.tls.your-tls-config.trust-store.pem.certs=ca.crt,ca2.pem
+# ...
+mp.messaging.incoming.prices.tls-configuration-name=your-tls-config
+----
+
 == Health reporting
 
 If you use the AMQP connector with the `quarkus-smallrye-health` extension, it contributes to the readiness and liveness probes.

docs/src/main/asciidoc/kafka.adoc

@@ -2153,6 +2153,36 @@ Update the `oauth.client.id`, `oauth.client.secret` and `oauth.token.endpoint.ur
 
 OAuth authentication works for both JVM and native modes. Since SSL in not enabled by default in native mode, `quarkus.ssl.native=true` must be added to support JaasClientOauthLoginCallbackHandler, which uses SSL. (See the xref:native-and-ssl.adoc[Using SSL with Native Executables] guide for more details.)
 
+== TLS Configuration
+
+Kafka client extension integrates with the xref:./tls-registry-reference.adoc[Quarkus TLS registry] to configure clients.
+
+To configure the TLS for the default Kafka configuration, you need to provide a named TLS configuration in the `application.properties`:
+
+[source, properties]
+----
+quarkus.tls.your-tls-config.trust-store.pem.certs=target/certs/kafka.crt,target/certs/kafka-ca.crt
+# ...
+kafka.tls-configuration-name=your-tls-config
+# enable ssl security protocol
+kafka.security.protocol=ssl
+----
+
+This will in turn provide the Kafka client with a `ssl.engine.factory.class` implementation.
+
+[IMPORTANT]
+====
+Make sure also to enable the SSL channel security protocol using the `security.protocol` property configured to `SSL` or `SASL_SSL`.
+====
+
+Quarkus Messaging channels can be configured individually to use a specific TLS configuration:
+
+[source, properties]
+----
+mp.messaging.incoming.your-channel.tls-configuration-name=your-tls-config
+mp.messaging.incoming.your-channel.security.protocol=ssl
+----
+
 == Testing a Kafka application
 
 === Testing without a broker

docs/src/main/asciidoc/messaging.adoc

@@ -600,6 +600,33 @@ You can disable tracing for a specific channel using the following configuration
 mp.messaging.incoming.data.tracing-enabled=false
 ----
 
+== TLS Configuration
+
+Some messaging extensions integrate with the xref:./tls-registry-reference.adoc[Quarkus TLS Registry] to configure the underlying client.
+To configure the TLS on a channel, you need to provide the named TLS configuration to the `tls-configuration-name` property:
+
+[source, properties]
+----
+quarkus.tls.my-tls-config.trust-store=truststore.jks
+quarkus.tls.my-tls-config.trust-store-password=secret
+mp.messaging.incoming.my-channel.tls-configuration-name=my-tls-config
+----
+
+Or you can configure it globally on all channels of a connector:
+
+[source, properties]
+----
+mp.messaging.connector.smallrye-pulsar.tls-configuration-name=my-tls-config
+----
+
+At the time of writing, the following messaging extensions support configuration through the Quarkus TLS Registry:
+
+* Kafka: Provides the `ssl.engine.factory.class` property for the Kafka client.
+* Pulsar: Only mTLS authentication is supported.
+* RabbitMQ
+* AMQP 1.0
+* MQTT
+
 == Testing
 
 === Testing with Dev Services

docs/src/main/asciidoc/pulsar.adoc

@@ -1118,6 +1118,23 @@ class PulsarConfig {
 }
 ----
 
+==== Configuring authentication to Pulsar using mTLS
+
+Pulsar Messaging extension integrates with the xref:./tls-registry-reference.adoc[Quarkus TLS registry] to authenticate clients using mTLS.
+
+To configure the mTLS for a Pulsar channel, you need to provide a named TLS configuration in the `application.properties`:
+
+[source, properties]
+----
+quarkus.tls.my-tls-config.trust-store.p12.path=target/certs/pulsar-client-truststore.p12
+quarkus.tls.my-tls-config.trust-store.p12.password=secret
+quarkus.tls.my-tls-config.key-store.p12.path=target/certs/pulsar-client-keystore.p12
+quarkus.tls.my-tls-config.key-store.p12.password=secret
+
+mp.messaging.incoming.prices.tls-configuration-name=my-tls-config
+----
+
+
 ==== Configuring access to Datastax Luna Streaming
 
 Luna Streaming is a production-ready distribution of Apache Pulsar, with tools and support from DataStax.

docs/src/main/asciidoc/rabbitmq-reference.adoc

@@ -389,6 +389,19 @@ You need to indicate the name of the client using the `client-options-name` attr
 mp.messaging.incoming.prices.client-options-name=my-named-options
 ----
 
+== TLS Configuration
+
+RabbitMQ Messaging extension integrates with the xref:./tls-registry-reference.adoc[Quarkus TLS registry] to configure the Vert.x RabbitMQ client.
+
+To configure the TLS for a channel, you need to provide a named TLS configuration in the `application.properties`:
+
+[source, properties]
+----
+quarkus.tls.your-tls-config.trust-store.pem.certs=ca.crt,ca2.pem
+# ...
+mp.messaging.incoming.prices.tls-configuration-name=your-tls-config
+----
+
 == Health reporting
 
 If you use the RabbitMQ connector with the `quarkus-smallrye-health` extension, it contributes to the readiness and liveness probes.

extensions/kafka-client/deployment/src/main/java/io/quarkus/kafka/client/deployment/KafkaProcessor.java

@@ -69,6 +69,7 @@
 import io.quarkus.deployment.annotations.Record;
 import io.quarkus.deployment.builditem.AdditionalIndexedClassesBuildItem;
 import io.quarkus.deployment.builditem.CombinedIndexBuildItem;
+import io.quarkus.deployment.builditem.ConfigDescriptionBuildItem;
 import io.quarkus.deployment.builditem.ExtensionSslNativeSupportBuildItem;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
 import io.quarkus.deployment.builditem.IndexDependencyBuildItem;
@@ -105,6 +106,7 @@
 import io.quarkus.kafka.client.serialization.ObjectMapperDeserializer;
 import io.quarkus.kafka.client.serialization.ObjectMapperSerializer;
 import io.quarkus.kafka.client.tls.QuarkusKafkaSslEngineFactory;
+import io.quarkus.runtime.annotations.ConfigPhase;
 import io.quarkus.smallrye.health.deployment.spi.HealthBuildItem;
 
 public class KafkaProcessor {
@@ -214,6 +216,7 @@ void relaxSaslElytron(BuildProducer<RunTimeConfigurationDefaultBuildItem> config
     @BuildStep
     public void build(
             KafkaBuildTimeConfig config, CurateOutcomeBuildItem curateOutcomeBuildItem,
+            BuildProducer<ConfigDescriptionBuildItem> configDescBuildItems,
             CombinedIndexBuildItem indexBuildItem, BuildProducer<ReflectiveClassBuildItem> reflectiveClass,
             BuildProducer<ServiceProviderBuildItem> serviceProviders,
             BuildProducer<NativeImageProxyDefinitionBuildItem> proxies,
@@ -289,6 +292,8 @@ public void build(
 
         reflectiveClass.produce(
                 ReflectiveClassBuildItem.builder(QuarkusKafkaSslEngineFactory.class).build());
+        configDescBuildItems.produce(new ConfigDescriptionBuildItem("kafka.tls-configuration-name", null,
+                "The tls-configuration to use for the Kafka client", null, null, ConfigPhase.RUN_TIME));
     }
 
     @BuildStep(onlyIf = { HasSnappy.class, NativeOrNativeSourcesBuild.class })

extensions/smallrye-reactive-messaging/deployment/src/main/java/io/quarkus/smallrye/reactivemessaging/deployment/WiringProcessor.java

@@ -214,6 +214,9 @@ private void handleMethodAnnotationWithOutgoing(BuildProducer<ChannelBuildItem>
                     new DeploymentException("Empty @Outgoing annotation on method " + method)));
         }
         if (outgoing != null) {
+            configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
+                    "mp.messaging.outgoing." + outgoing.value().asString() + ".tls-configuration-name", null,
+                    "The tls-configuration to use", null, null, ConfigPhase.RUN_TIME));
             configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
                     "mp.messaging.outgoing." + outgoing.value().asString() + ".connector", null,
                     "The connector to use", null, null, ConfigPhase.BUILD_TIME));
@@ -232,6 +235,9 @@ private void handleMethodAnnotationWithOutgoings(BuildProducer<ChannelBuildItem>
                     validationErrors.produce(new ValidationPhaseBuildItem.ValidationErrorBuildItem(
                             new DeploymentException("Empty @Outgoing annotation on method " + method)));
                 }
+                configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
+                        "mp.messaging.outgoing." + instance.value().asString() + ".tls-configuration-name", null,
+                        "The tls-configuration to use", null, null, ConfigPhase.RUN_TIME));
                 configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
                         "mp.messaging.outgoing." + instance.value().asString() + ".connector", null,
                         "The connector to use", null, null, ConfigPhase.BUILD_TIME));
@@ -250,6 +256,9 @@ private void handleMethodAnnotationWithIncomings(BuildProducer<ChannelBuildItem>
                     validationErrors.produce(new ValidationPhaseBuildItem.ValidationErrorBuildItem(
                             new DeploymentException("Empty @Incoming annotation on method " + method)));
                 }
+                configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
+                        "mp.messaging.incoming." + instance.value().asString() + ".tls-configuration-name", null,
+                        "The tls-configuration to use", null, null, ConfigPhase.RUN_TIME));
                 configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
                         "mp.messaging.incoming." + instance.value().asString() + ".connector", null,
                         "The connector to use", null, null, ConfigPhase.BUILD_TIME));
@@ -267,6 +276,9 @@ private void handleMethodAnnotatedWithIncoming(BuildProducer<ChannelBuildItem> a
                     new DeploymentException("Empty @Incoming annotation on method " + method)));
         }
         if (incoming != null) {
+            configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
+                    "mp.messaging.incoming." + incoming.value().asString() + ".tls-configuration-name", null,
+                    "The tls-configuration to use", null, null, ConfigPhase.RUN_TIME));
             configDescriptionBuildItemBuildProducer.produce(new ConfigDescriptionBuildItem(
                     "mp.messaging.incoming." + incoming.value().asString() + ".connector", null,
                     "The connector to use", null, null, ConfigPhase.BUILD_TIME));