Repository: zeppelin Updated Branches: refs/heads/master b183c7e06 -> 733bed591
ZEPPELIN-3818. Unit test failed: ZeppelinSparkClusterTest.testAngularObjects ### What is this PR for? This is to fix the unit test failure of ZeppelinSparkClusterTest.testAngularObjects. It is due to the parsing logic bug in RemoteInterpreterUtils.java. Besides the fix, I also did some refactoring to add more logging. ### What type of PR is it? [Bug Fix | Refactoring] ### Todos * [ ] - Task ### What is the Jira issue? * https://jira.apache.org/jira/browse/ZEPPELIN-3818 ### How should this be tested? * CI pass ### Screenshots (if appropriate) ### Questions: * Does the licenses files need update? No * Is there breaking changes for older versions? NO * Does this needs documentation? NO Author: Jeff Zhang <zjf...@apache.org> Closes #3209 from zjffdu/ZEPPELIN-3818 and squashes the following commits: 8d462964e [Jeff Zhang] ZEPPELIN-3818. Unit test failed: ZeppelinSparkClusterTest.testAngularObjects Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/733bed59 Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/733bed59 Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/733bed59 Branch: refs/heads/master Commit: 733bed591bdb6a06864331540364254f1ea1ff20 Parents: b183c7e Author: Jeff Zhang <zjf...@apache.org> Authored: Fri Oct 26 14:58:21 2018 +0800 Committer: Jeff Zhang <zjf...@apache.org> Committed: Fri Oct 26 16:08:59 2018 +0800 ---------------------------------------------------------------------- .../remote/RemoteInterpreterServer.java | 3 ++- .../remote/RemoteInterpreterUtils.java | 2 +- .../RemoteInterpreterEventServer.java | 24 ++++---------------- .../zeppelin/scheduler/RemoteScheduler.java | 1 - 4 files changed, 7 insertions(+), 23 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/zeppelin/blob/733bed59/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java ---------------------------------------------------------------------- diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java index e588e40..dde73a7 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterServer.java @@ -142,6 +142,8 @@ public class RemoteInterpreterServer extends Thread String interpreterGroupId, boolean isTest) throws TTransportException, IOException { + logger.info("Starting remote interpreter server on port {}, intpEventServerAddress: {}:{}", port, + intpEventServerHost, intpEventServerPort); if (null != intpEventServerHost) { this.intpEventServerHost = intpEventServerHost; if (!isTest) { @@ -171,7 +173,6 @@ public class RemoteInterpreterServer extends Thread } server = new TThreadPoolServer( new TThreadPoolServer.Args(serverTransport).processor(processor)); - logger.info("Starting remote interpreter server on port {}", port); remoteWorksResponsePool = Collections.synchronizedMap(new HashMap<String, Object>()); } http://git-wip-us.apache.org/repos/asf/zeppelin/blob/733bed59/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtils.java ---------------------------------------------------------------------- diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtils.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtils.java index cf82247..7e9ff13 100644 --- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtils.java +++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/interpreter/remote/RemoteInterpreterUtils.java @@ -138,7 +138,7 @@ public class RemoteInterpreterUtils { public static String getInterpreterSettingId(String intpGrpId) { String settingId = null; if (intpGrpId != null) { - int indexOfColon = intpGrpId.indexOf(":"); + int indexOfColon = intpGrpId.indexOf("-"); settingId = intpGrpId.substring(0, indexOfColon); } return settingId; http://git-wip-us.apache.org/repos/asf/zeppelin/blob/733bed59/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java index 5b7223c..bd612d6 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/RemoteInterpreterEventServer.java @@ -87,24 +87,6 @@ public class RemoteInterpreterEventServer implements RemoteInterpreterEventServi this.appListener = interpreterSettingManager.getAppEventListener(); } - public void run() { - TServerSocket tSocket = null; - try { - tSocket = RemoteInterpreterUtils.createTServerSocket(portRange); - port = tSocket.getServerSocket().getLocalPort(); - host = RemoteInterpreterUtils.findAvailableHostAddress(); - } catch (IOException e1) { - throw new RuntimeException(e1); - } - - LOGGER.info("InterpreterEventServer will start. Port: {}", port); - RemoteInterpreterEventService.Processor processor = - new RemoteInterpreterEventService.Processor(this); - this.thriftServer = new TThreadPoolServer( - new TThreadPoolServer.Args(tSocket).processor(processor)); - this.thriftServer.serve(); - } - public void start() throws IOException { Thread startingThread = new Thread() { @Override @@ -118,7 +100,7 @@ public class RemoteInterpreterEventServer implements RemoteInterpreterEventServi throw new RuntimeException(e1); } - LOGGER.info("InterpreterEventServer will start. Port: {}", port); + LOGGER.info("InterpreterEventServer is starting at {}:{}", host, port); RemoteInterpreterEventService.Processor processor = new RemoteInterpreterEventService.Processor(RemoteInterpreterEventServer.this); thriftServer = new TThreadPoolServer( @@ -142,7 +124,7 @@ public class RemoteInterpreterEventServer implements RemoteInterpreterEventServi if (thriftServer != null && !thriftServer.isServing()) { throw new IOException("Fail to start InterpreterEventServer in 30 seconds."); } - LOGGER.info("InterpreterEventServer is started"); + LOGGER.info("RemoteInterpreterEventServer is started"); runner = new AppendOutputRunner(listener); appendFuture = appendService.scheduleWithFixedDelay( @@ -156,6 +138,7 @@ public class RemoteInterpreterEventServer implements RemoteInterpreterEventServi if (appendFuture != null) { appendFuture.cancel(true); } + LOGGER.info("RemoteInterpreterEventServer is stopped"); } @@ -252,6 +235,7 @@ public class RemoteInterpreterEventServer implements RemoteInterpreterEventServi @Override public void addAngularObject(String intpGroupId, String json) throws TException { + LOGGER.debug("Add AngularObject, interpreterGroupId: " + intpGroupId + ", json: " + json); AngularObject angularObject = AngularObject.fromJson(json); InterpreterGroup interpreterGroup = interpreterSettingManager.getInterpreterGroupById(intpGroupId); http://git-wip-us.apache.org/repos/asf/zeppelin/blob/733bed59/zeppelin-zengine/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java ---------------------------------------------------------------------- diff --git a/zeppelin-zengine/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java b/zeppelin-zengine/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java index 177dfec..34f1d5d 100644 --- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java +++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java @@ -147,7 +147,6 @@ public class RemoteScheduler extends AbstractScheduler { public synchronized Status getStatus() { if (!remoteInterpreter.isOpened()) { - LOGGER.info("status:" + getLastStatus()); return getLastStatus(); } Status status = Status.valueOf(remoteInterpreter.getStatus(job.getId()));