emit Pulsar source and sink configuration property values to logs #335

Author: gavingaozhangmin

pulsar-flink-connector/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() {

pulsar-flink-connector/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.Iterator;
@@ -496,6 +499,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() {