Adding logging to JavaInstanceRunnable to help debug issues like #10460…

… (#10461) As reported in #10460 (and for other purposes), it's helpful to see the context when a function crashes when trying to open its sink. This PR adds a couple of debug lines to assist in troubleshooting those cases.
apache · May 4, 2021 · 06bb15b · 06bb15b
1 parent f3f1f85
commit 06bb15b
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 0 deletions.
diff --git a/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java b/...ar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/ContextImpl.java
@@ -35,6 +35,8 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+
+import lombok.ToString;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.CompressionType;
@@ -75,6 +77,7 @@
 /**
  * This class implements the Context interface exposed to the user.
  */
+@ToString
 class ContextImpl implements Context, SinkContext, SourceContext, AutoCloseable {
     private InstanceConfig config;
     private Logger logger;

diff --git a/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -796,8 +796,15 @@ private void setupOutput(ContextImpl contextImpl) throws Exception {
         }
         try {
             if (sinkSpec.getConfigs().isEmpty()) {
+                if (log.isDebugEnabled()){
+                    log.debug("Opening Sink with empty hashmap with contextImpl: {} ", contextImpl.toString());
+                }
                 this.sink.open(new HashMap<>(), contextImpl);
             } else {
+                if (log.isDebugEnabled()){
+                    log.debug("Opening Sink with SinkSpec {} and contextImpl: {} ", sinkSpec.toString(),
+                            contextImpl.toString());
+                }
                 this.sink.open(ObjectMapperFactory.getThreadLocal().readValue(sinkSpec.getConfigs(),
                         new TypeReference<Map<String, Object>>() {}), contextImpl);
             }