emit Pulsar source and sink configuration property values to logs str…

…eamnative#335
gaozhangmin · Aug 30, 2021 · 9453349 · 9453349
1 parent 0280311
commit 9453349
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 0 deletions.
diff --git a/...ector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSinkBase.java b/...ector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSinkBase.java
@@ -40,6 +40,8 @@
 
 import org.apache.flink.shaded.guava18.com.google.common.collect.Maps;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.MessageId;
@@ -271,6 +273,14 @@ public void open(Configuration parameters) throws Exception {
             topic2Producer = new HashMap<>();
         }
        //super.open(parameters);
+
+        try {
+            ObjectMapper m = new ObjectMapper();
+            ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+            log.info("Pulsar sink config: {}", w.writeValueAsString(properties));
+        } catch (IOException e) {
+            log.error("Failed to dump sink config info", e);
+        }
     }
 
     protected void initializeSendCallback() {

diff --git a/...nnector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java b/...nnector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java
@@ -62,13 +62,16 @@
 
 import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.apache.pulsar.shade.com.google.common.collect.Maps;
 import org.apache.pulsar.shade.org.apache.commons.lang3.StringUtils;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -501,6 +504,14 @@ public void open(Configuration parameters) throws Exception {
                         taskIndex, ownedTopicStarts.size(), ownedTopicStarts);
             }
         }
+
+        try {
+            ObjectMapper m = new ObjectMapper();
+            ObjectWriter w = m.writerWithDefaultPrettyPrinter();
+            log.info("Pulsar source config: {}", w.writeValueAsString(properties));
+        } catch (IOException e) {
+            log.error("Failed to dump source config info", e);
+        }
     }
 
     protected String getSubscriptionName() {