add isinfoEnabled

Author: gavingaozhangmin

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSinkBase.java

@@ -273,13 +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);
+        if (log.isInfoEnabled()) {
+            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);
+            }
         }
     }
 

pulsar-flink-connector/src/main/java/org/apache/flink/streaming/connectors/pulsar/FlinkPulsarSource.java

@@ -505,12 +505,14 @@ public void open(Configuration parameters) throws Exception {
             }
         }
 
-        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);
+        if (log.isInfoEnabled()) {
+            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);
+            }
         }
     }