From 66d3d2fef536f27d6e21bf74f901618c021c0aa7 Mon Sep 17 00:00:00 2001 From: astroshim Date: Fri, 20 Jan 2017 13:41:01 -0800 Subject: [PATCH 1/2] fix to thread safe --- .../apache/zeppelin/interpreter/remote/AppendOutputRunner.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java index 03d919187e4..85102b2e509 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java @@ -41,7 +41,7 @@ public class AppendOutputRunner implements Runnable { private static final Long SAFE_PROCESSING_TIME = new Long(10); private static final Long SAFE_PROCESSING_STRING_SIZE = new Long(100000); - private final BlockingQueue queue = new LinkedBlockingQueue<>(); + private static final BlockingQueue queue = new LinkedBlockingQueue<>(); private final RemoteInterpreterProcessListener listener; public AppendOutputRunner(RemoteInterpreterProcessListener listener) { From e99358541c23bd0cefe5e3dbd2eb7d96890a0ab0 Mon Sep 17 00:00:00 2001 From: astroshim Date: Fri, 20 Jan 2017 16:18:40 -0800 Subject: [PATCH 2/2] make scheduler not static --- .../apache/zeppelin/interpreter/remote/AppendOutputRunner.java | 2 +- .../interpreter/remote/RemoteInterpreterEventPoller.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java index 85102b2e509..03d919187e4 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/AppendOutputRunner.java @@ -41,7 +41,7 @@ public class AppendOutputRunner implements Runnable { private static final Long SAFE_PROCESSING_TIME = new Long(10); private static final Long SAFE_PROCESSING_STRING_SIZE = new Long(100000); - private static final BlockingQueue queue = new LinkedBlockingQueue<>(); + private final BlockingQueue queue = new LinkedBlockingQueue<>(); private final RemoteInterpreterProcessListener listener; public AppendOutputRunner(RemoteInterpreterProcessListener listener) { diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java index bf64d9fa084..e2a8adddfd2 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterEventPoller.java @@ -52,7 +52,7 @@ */ public class RemoteInterpreterEventPoller extends Thread { private static final Logger logger = LoggerFactory.getLogger(RemoteInterpreterEventPoller.class); - private static final ScheduledExecutorService appendService = + private final ScheduledExecutorService appendService = Executors.newSingleThreadScheduledExecutor(); private final RemoteInterpreterProcessListener listener; private final ApplicationEventListener appListener;