Fix Pulsar log messages to include channel-specific configuration

Fixes #3154

Author: ozangunalp

smallrye-reactive-messaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/ConfigResolver.java

@@ -24,7 +24,9 @@
 import org.apache.pulsar.client.api.RedeliveryBackoff;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.ClientBuilderImpl;
+import org.apache.pulsar.client.impl.ConsumerBuilderImpl;
 import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff;
+import org.apache.pulsar.client.impl.ProducerBuilderImpl;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
 import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
@@ -356,4 +358,12 @@ private static boolean allCaps(String key) {
         return key.toUpperCase().equals(key);
     }
 
+    public ConsumerConfigurationData<?> getConfig(ConsumerBuilder<?> consumerBuilder) {
+        return ((ConsumerBuilderImpl<?>) consumerBuilder).getConf();
+    }
+
+    public ProducerConfigurationData getConfig(ProducerBuilder<?> producerBuilder) {
+        return ((ProducerBuilderImpl<?>) producerBuilder).getConf();
+    }
+
 }

smallrye-reactive-messaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/PulsarIncomingChannel.java

@@ -75,7 +75,7 @@ public PulsarIncomingChannel(PulsarClient client, Vertx vertx, Schema<T> schema,
         }
         ConsumerBuilder<T> builder = configResolver.configure(client.newConsumer(schema), ic, conf);
         this.consumer = builder.subscribe();
-        log.createdConsumerWithConfig(channel, SchemaResolver.getSchemaName(schema), conf);
+        log.createdConsumerWithConfig(channel, SchemaResolver.getSchemaName(schema), configResolver.getConfig(builder));
         this.ackHandler = ackHandlerFactory.create(consumer, ic);
         this.failureHandler = failureHandlerFactory.create(consumer, ic, this::reportFailure);
         this.context = ((VertxInternal) vertx.getDelegate()).createEventLoopContext();

smallrye-reactive-messaging-pulsar/src/main/java/io/smallrye/reactive/messaging/pulsar/PulsarOutgoingChannel.java

@@ -59,7 +59,7 @@ public PulsarOutgoingChannel(PulsarClient client, Schema<T> schema, PulsarConnec
         }
         ProducerBuilder<T> builder = configResolver.configure(client.newProducer(schema), oc, conf);
         this.producer = builder.create();
-        log.createdProducerWithConfig(channel, SchemaResolver.getSchemaName(schema), conf);
+        log.createdProducerWithConfig(channel, SchemaResolver.getSchemaName(schema), configResolver.getConfig(builder));
         long requests = getRequests(oc, conf);
 
         processor = new SenderProcessor(requests, oc.getWaitForWriteCompletion(), this::sendMessage);