[improve][io] Enhance Kafka connector logging with focused bootstrap server information (#24128)

freeznet · lhotari · commit 5b1a6430e67a · 2025-03-26T16:50:34.000+02:00
(cherry picked from commit d5f126f)
diff --git a/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java b/pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSink.java
@@ -44,6 +44,7 @@
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
+import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.connect.connector.Task;
@@ -160,7 +161,7 @@ public void open(Map<String, Object> config, SinkContext ctx) throws Exception {
         kafkaSinkConfig = PulsarKafkaConnectSinkConfig.load(config);
         Objects.requireNonNull(kafkaSinkConfig.getTopic(), "Kafka topic is not set");
         Preconditions.checkArgument(ctx.getSubscriptionType() == SubscriptionType.Failover
-                || ctx.getSubscriptionType() == SubscriptionType.Exclusive,
+                        || ctx.getSubscriptionType() == SubscriptionType.Exclusive,
                 "Source must run with Exclusive or Failover subscription type");
         topicName = kafkaSinkConfig.getTopic();
         unwrapKeyValueIfAvailable = kafkaSinkConfig.isUnwrapKeyValueIfAvailable();
@@ -220,7 +221,7 @@ public void open(Map<String, Object> config, SinkContext ctx) throws Exception {
         scheduledExecutor.scheduleWithFixedDelay(() ->
                 this.flushIfNeeded(true), lingerMs, lingerMs, TimeUnit.MILLISECONDS);
 
-        log.info("Kafka sink started : {}.", props);
+        log.info("Kafka sink started : {}.", props.getOrDefault(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ""));
     }
 
     private void flushIfNeeded(boolean force) {
@@ -282,15 +283,15 @@ private static boolean areMapsEqual(Map<TopicPartition, OffsetAndMetadata> first
 
     @VisibleForTesting
     protected void ackUntil(Record<GenericObject> lastNotFlushed,
-                          Map<TopicPartition, OffsetAndMetadata> committedOffsets,
-                          java.util.function.Consumer<Record<GenericObject>> cb) {
+                            Map<TopicPartition, OffsetAndMetadata> committedOffsets,
+                            java.util.function.Consumer<Record<GenericObject>> cb) {
         // lastNotFlushed is needed in case of default preCommit() implementation
         // which calls flush() and returns currentOffsets passed to it.
         // We don't want to ack messages added to pendingFlushQueue after the preCommit/flush call
 
         // to avoid creation of new TopicPartition for each record in pendingFlushQueue
         Map<String, Map<Integer, Long>> topicOffsets = new HashMap<>();
-        for (Map.Entry<TopicPartition, OffsetAndMetadata> e: committedOffsets.entrySet()) {
+        for (Map.Entry<TopicPartition, OffsetAndMetadata> e : committedOffsets.entrySet()) {
             TopicPartition tp = e.getKey();
             if (!topicOffsets.containsKey(tp.topic())) {
                 topicOffsets.put(tp.topic(), new HashMap<>());
diff --git a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaAbstractSink.java
@@ -81,11 +81,11 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
         kafkaSinkConfig = KafkaSinkConfig.load(config, sinkContext);
         if (kafkaSinkConfig.getBatchSize() <= 0) {
             throw new IllegalArgumentException("Invalid Kafka Producer batchSize : "
-                + kafkaSinkConfig.getBatchSize());
+                    + kafkaSinkConfig.getBatchSize());
         }
         if (kafkaSinkConfig.getMaxRequestSize() <= 0) {
             throw new IllegalArgumentException("Invalid Kafka Producer maxRequestSize : "
-                + kafkaSinkConfig.getMaxRequestSize());
+                    + kafkaSinkConfig.getMaxRequestSize());
         }
         if (kafkaSinkConfig.getProducerConfigProperties() != null) {
             props.putAll(kafkaSinkConfig.getProducerConfigProperties());
@@ -122,7 +122,7 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc
 
         producer = new KafkaProducer<>(beforeCreateProducer(props));
 
-        log.info("Kafka sink started : {}.", props);
+        log.info("Kafka sink started : {}.", props.getOrDefault(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ""));
     }
 
     public abstract KeyValue<K, V> extractKeyValue(Record<byte[]> message);
diff --git a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSink.java b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSink.java
@@ -33,10 +33,10 @@
  * apply schema into it.
  */
 @Connector(
-    name = "kafka",
-    type = IOType.SINK,
-    help = "The KafkaBytesSink is used for moving messages from Pulsar to Kafka.",
-    configClass = KafkaSinkConfig.class
+        name = "kafka",
+        type = IOType.SINK,
+        help = "The KafkaBytesSink is used for moving messages from Pulsar to Kafka.",
+        configClass = KafkaSinkConfig.class
 )
 @Slf4j
 public class KafkaBytesSink extends KafkaAbstractSink<String, byte[]> {
@@ -45,7 +45,7 @@ public class KafkaBytesSink extends KafkaAbstractSink<String, byte[]> {
     protected Properties beforeCreateProducer(Properties props) {
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class.getName());
-        log.info("Created kafka producer config : {}", props);
+        log.info("Created kafka producer on : {}", props.getOrDefault(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ""));
         return props;
     }
 
diff --git a/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java b/pulsar-io/kafka/src/main/java/org/apache/pulsar/io/kafka/KafkaBytesSource.java
@@ -49,22 +49,22 @@
 import org.apache.pulsar.io.core.annotations.IOType;
 
 /**
- *  Kafka Source that transfers the data from Kafka to Pulsar and sets the Schema type properly.
- *  We use the key and the value deserializer in order to decide the type of Schema to be set on the topic on Pulsar.
- *  In case of KafkaAvroDeserializer we use the Schema Registry to download the schema and apply it to the topic.
- *  Please refer to {@link #getSchemaFromDeserializerAndAdaptConfiguration(String, Properties, boolean)} for the list
- *  of supported Deserializers.
- *  If you set StringDeserializer for the key then we use the raw key as key for the Pulsar message.
- *  If you set another Deserializer for the key we use the KeyValue schema type in Pulsar with the SEPARATED encoding.
- *  This way the Key is stored in the Pulsar key, encoded as base64 string and with a Schema, the Value of the message
- *  is stored in the Pulsar value with a Schema.
- *  This way there is a one-to-one mapping between Kafka key/value pair and the Pulsar data model.
+ * Kafka Source that transfers the data from Kafka to Pulsar and sets the Schema type properly.
+ * We use the key and the value deserializer in order to decide the type of Schema to be set on the topic on Pulsar.
+ * In case of KafkaAvroDeserializer we use the Schema Registry to download the schema and apply it to the topic.
+ * Please refer to {@link #getSchemaFromDeserializerAndAdaptConfiguration(String, Properties, boolean)} for the list
+ * of supported Deserializers.
+ * If you set StringDeserializer for the key then we use the raw key as key for the Pulsar message.
+ * If you set another Deserializer for the key we use the KeyValue schema type in Pulsar with the SEPARATED encoding.
+ * This way the Key is stored in the Pulsar key, encoded as base64 string and with a Schema, the Value of the message
+ * is stored in the Pulsar value with a Schema.
+ * This way there is a one-to-one mapping between Kafka key/value pair and the Pulsar data model.
  */
 @Connector(
-    name = "kafka",
-    type = IOType.SOURCE,
-    help = "Transfer data from Kafka to Pulsar.",
-    configClass = KafkaSourceConfig.class
+        name = "kafka",
+        type = IOType.SOURCE,
+        help = "Transfer data from Kafka to Pulsar.",
+        configClass = KafkaSourceConfig.class
 )
 @Slf4j
 public class KafkaBytesSource extends KafkaAbstractSource<ByteBuffer> {
@@ -78,15 +78,15 @@ public class KafkaBytesSource extends KafkaAbstractSource<ByteBuffer> {
     protected Properties beforeCreateConsumer(Properties props) {
         props.putIfAbsent(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
         props.putIfAbsent(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
-        log.info("Created kafka consumer config : {}", props);
+        log.info("Created kafka consumer on : {}", props.getOrDefault(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ""));
 
         keySchema = getSchemaFromDeserializerAndAdaptConfiguration(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                 props, true);
         valueSchema = getSchemaFromDeserializerAndAdaptConfiguration(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                 props, false);
 
         boolean needsSchemaCache = keySchema == DeferredSchemaPlaceholder.INSTANCE
-                                    || valueSchema == DeferredSchemaPlaceholder.INSTANCE;
+                || valueSchema == DeferredSchemaPlaceholder.INSTANCE;
 
         if (needsSchemaCache) {
             initSchemaCache(props);
@@ -174,8 +174,8 @@ private static Schema<ByteBuffer> getSchemaFromDeserializerAndAdaptConfiguration
 
         Schema<?> result;
         if (ByteArrayDeserializer.class.getName().equals(kafkaDeserializerClass)
-            || ByteBufferDeserializer.class.getName().equals(kafkaDeserializerClass)
-            || BytesDeserializer.class.getName().equals(kafkaDeserializerClass)) {
+                || ByteBufferDeserializer.class.getName().equals(kafkaDeserializerClass)
+                || BytesDeserializer.class.getName().equals(kafkaDeserializerClass)) {
             result = Schema.BYTEBUFFER;
         } else if (StringDeserializer.class.getName().equals(kafkaDeserializerClass)) {
             if (isKey) {
@@ -193,7 +193,7 @@ private static Schema<ByteBuffer> getSchemaFromDeserializerAndAdaptConfiguration
             result = Schema.INT64;
         } else if (ShortDeserializer.class.getName().equals(kafkaDeserializerClass)) {
             result = Schema.INT16;
-        } else if (KafkaAvroDeserializer.class.getName().equals(kafkaDeserializerClass)){
+        } else if (KafkaAvroDeserializer.class.getName().equals(kafkaDeserializerClass)) {
             // in this case we have to inject our custom deserializer
             // that extracts Avro schema information
             props.put(key, ExtractKafkaAvroSchemaDeserializer.class.getName());
@@ -238,7 +238,7 @@ public BytesWithKafkaSchema deserialize(String topic, byte[] payload) {
         }
     }
 
-     static final class DeferredSchemaPlaceholder extends ByteBufferSchemaWrapper {
+    static final class DeferredSchemaPlaceholder extends ByteBufferSchemaWrapper {
         DeferredSchemaPlaceholder() {
             super(SchemaInfoImpl
                     .builder()
@@ -247,6 +247,7 @@ static final class DeferredSchemaPlaceholder extends ByteBufferSchemaWrapper {
                     .schema(new byte[0])
                     .build());
         }
+
         static final DeferredSchemaPlaceholder INSTANCE = new DeferredSchemaPlaceholder();
     }
 

Original file line number	Diff line number	Diff line change
`@@ -81,11 +81,11 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc`
`81`	`81`	`kafkaSinkConfig = KafkaSinkConfig.load(config, sinkContext);`
`82`	`82`	`if (kafkaSinkConfig.getBatchSize() <= 0) {`
`83`	`83`	`throw new IllegalArgumentException("Invalid Kafka Producer batchSize : "`
`84`		`- + kafkaSinkConfig.getBatchSize());`
	`84`	`+ + kafkaSinkConfig.getBatchSize());`
`85`	`85`	`}`
`86`	`86`	`if (kafkaSinkConfig.getMaxRequestSize() <= 0) {`
`87`	`87`	`throw new IllegalArgumentException("Invalid Kafka Producer maxRequestSize : "`
`88`		`- + kafkaSinkConfig.getMaxRequestSize());`
	`88`	`+ + kafkaSinkConfig.getMaxRequestSize());`
`89`	`89`	`}`
`90`	`90`	`if (kafkaSinkConfig.getProducerConfigProperties() != null) {`
`91`	`91`	`props.putAll(kafkaSinkConfig.getProducerConfigProperties());`
`@@ -122,7 +122,7 @@ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exc`
`122`	`122`
`123`	`123`	`producer = new KafkaProducer<>(beforeCreateProducer(props));`
`124`	`124`
`125`		`- log.info("Kafka sink started : {}.", props);`
	`125`	`+ log.info("Kafka sink started : {}.", props.getOrDefault(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ""));`
`126`	`126`	`}`
`127`	`127`
`128`	`128`	`public abstract KeyValue<K, V> extractKeyValue(Record<byte[]> message);`