RemoteInterpreter extends Interpreter { @Override public InterpreterResult interpret(final String st, final InterpreterContext context) throws InterpreterException { if (LOGGER.isDebugEnabled()) { LOGGER.debug("st:\n{}", st); } final FormType form = getFormType(); RemoteInterpreterProcess interpreterProcess = null; try { interpreterProcess = getOrCreateInterpreterProcess(); } catch (IOException e) { throw new InterpreterException(e); } if (!interpreterProcess.isRunning()) { return new InterpreterResult(InterpreterResult.Code.ERROR, "Interpreter process is not running\n" + interpreterProcess.getErrorMessage()); } return interpreterProcess.callRemoteFunction(client -> { RemoteInterpreterResult remoteResult = client.interpret( sessionId, className, st, convert(context)); Map<String, Object> remoteConfig = (Map<String, Object>) GSON.fromJson( remoteResult.getConfig(), new TypeToken<Map<String, Object>>() { }.getType()); context.getConfig().clear(); if (remoteConfig != null) { context.getConfig().putAll(remoteConfig); } GUI currentGUI = context.getGui(); GUI currentNoteGUI = context.getNoteGui(); if (form == FormType.NATIVE) { GUI remoteGui = GUI.fromJson(remoteResult.getGui()); GUI remoteNoteGui = GUI.fromJson(remoteResult.getNoteGui()); currentGUI.clear(); currentGUI.setParams(remoteGui.getParams()); currentGUI.setForms(remoteGui.getForms()); currentNoteGUI.setParams(remoteNoteGui.getParams()); currentNoteGUI.setForms(remoteNoteGui.getForms()); } else if (form == FormType.SIMPLE) { final Map<String, Input> currentForms = currentGUI.getForms(); final Map<String, Object> currentParams = currentGUI.getParams(); final GUI remoteGUI = GUI.fromJson(remoteResult.getGui()); final Map<String, Input> remoteForms = remoteGUI.getForms(); final Map<String, Object> remoteParams = remoteGUI.getParams(); currentForms.putAll(remoteForms); currentParams.putAll(remoteParams); } InterpreterResult result = convert(remoteResult); return result; } ); } RemoteInterpreter(Properties properties,
                           String sessionId,
                           String className,
                           String userName); boolean isOpened(); @VisibleForTesting void setOpened(boolean opened); @Override String getClassName(); String getSessionId(); synchronized RemoteInterpreterProcess getOrCreateInterpreterProcess(); ManagedInterpreterGroup getInterpreterGroup(); @Override void open(); @Override void close(); @Override InterpreterResult interpret(final String st, final InterpreterContext context); @Override void cancel(final InterpreterContext context); @Override FormType getFormType(); @Override int getProgress(final InterpreterContext context); @Override List<InterpreterCompletion> completion(final String buf, final int cursor,
                                                final InterpreterContext interpreterContext); String getStatus(final String jobId); @Override Scheduler getScheduler(); @Override String toString();  }
@Test public void testExecuteIncorrectPrecode() throws TTransportException, IOException, InterpreterException { interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED); interpreterSetting.setProperty("zeppelin.SleepInterpreter.precode", "fail test"); Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "sleep"); InterpreterContext context1 = createDummyInterpreterContext();; assertEquals(Code.ERROR, interpreter1.interpret("10", context1).code()); }
@Test public void testExecuteCorrectPrecode() throws TTransportException, IOException, InterpreterException { interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED); interpreterSetting.setProperty("zeppelin.SleepInterpreter.precode", "1"); Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "sleep"); InterpreterContext context1 = createDummyInterpreterContext(); assertEquals(Code.SUCCESS, interpreter1.interpret("10", context1).code()); }
@Test public void testRemoteInterperterErrorStatus() throws TTransportException, IOException, InterpreterException { interpreterSetting.setProperty("zeppelin.interpreter.echo.fail", "true"); interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED); Interpreter interpreter1 = interpreterSetting.getDefaultInterpreter("user1", "note1"); assertTrue(interpreter1 instanceof RemoteInterpreter); RemoteInterpreter remoteInterpreter1 = (RemoteInterpreter) interpreter1; InterpreterContext context1 = createDummyInterpreterContext();; assertEquals(Code.ERROR, remoteInterpreter1.interpret("hello", context1).code()); }
@Test public void testFIFOScheduler() throws InterruptedException, InterpreterException { interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED); final Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "sleep"); final InterpreterContext context1 = createDummyInterpreterContext(); interpreter1.interpret("1", context1); Thread thread1 = new Thread() { @Override public void run() { try { assertEquals(Code.SUCCESS, interpreter1.interpret("100", context1).code()); } catch (InterpreterException e) { e.printStackTrace(); fail(); } } }; Thread thread2 = new Thread() { @Override public void run() { try { assertEquals(Code.SUCCESS, interpreter1.interpret("100", context1).code()); } catch (InterpreterException e) { e.printStackTrace(); fail(); } } }; long start = System.currentTimeMillis(); thread1.start(); thread2.start(); thread1.join(); thread2.join(); long end = System.currentTimeMillis(); assertTrue((end - start) >= 200); }
@Test public void testParallelScheduler() throws InterruptedException, InterpreterException { interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED); interpreterSetting.setProperty("zeppelin.SleepInterpreter.parallel", "true"); final Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "sleep"); final InterpreterContext context1 = createDummyInterpreterContext(); interpreter1.interpret("1", context1); Thread thread1 = new Thread() { @Override public void run() { try { assertEquals(Code.SUCCESS, interpreter1.interpret("100", context1).code()); } catch (InterpreterException e) { e.printStackTrace(); fail(); } } }; Thread thread2 = new Thread() { @Override public void run() { try { assertEquals(Code.SUCCESS, interpreter1.interpret("100", context1).code()); } catch (InterpreterException e) { e.printStackTrace(); fail(); } } }; long start = System.currentTimeMillis(); thread1.start(); thread2.start(); thread1.join(); thread2.join(); long end = System.currentTimeMillis(); assertTrue((end - start) <= 200); }
@Test public void testEnvironmentAndProperty() throws InterpreterException { interpreterSetting.getOption().setPerUser(InterpreterOption.SHARED); interpreterSetting.setProperty("ENV_1", "VALUE_1"); interpreterSetting.setProperty("property_1", "value_1"); final Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "get"); final InterpreterContext context1 = createDummyInterpreterContext(); assertEquals("VALUE_1", interpreter1.interpret("getEnv ENV_1", context1).message().get(0).getData()); assertEquals("null", interpreter1.interpret("getEnv ENV_2", context1).message().get(0).getData()); assertEquals("value_1", interpreter1.interpret("getProperty property_1", context1).message().get(0).getData()); assertEquals("null", interpreter1.interpret("getProperty not_existed_property", context1).message().get(0).getData()); }
@Test public void testConvertDynamicForms() throws InterpreterException { GUI gui = new GUI(); OptionInput.ParamOption[] paramOptions = { new OptionInput.ParamOption("value1", "param1"), new OptionInput.ParamOption("value2", "param2") }; List<Object> defaultValues = new ArrayList(); defaultValues.add("default1"); defaultValues.add("default2"); gui.checkbox("checkbox_id", paramOptions, defaultValues); gui.select("select_id", paramOptions, "default"); gui.textbox("textbox_id"); Map<String, Input> expected = new LinkedHashMap<>(gui.getForms()); Interpreter interpreter = interpreterSetting.getDefaultInterpreter("user1", "note1"); InterpreterContext context = createDummyInterpreterContext(); interpreter.interpret("text", context); assertArrayEquals(expected.values().toArray(), gui.getForms().values().toArray()); }
@Test public void testFailToLaunchInterpreterProcess_InvalidRunner() { try { System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_REMOTE_RUNNER.getVarName(), "invalid_runner"); final Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "sleep"); final InterpreterContext context1 = createDummyInterpreterContext(); try { interpreter1.interpret("1", context1); fail("Should not be able to launch interpreter process"); } catch (InterpreterException e) { assertTrue(ExceptionUtils.getStackTrace(e).contains("java.io.IOException")); } } finally { System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_REMOTE_RUNNER.getVarName()); } }
@Test public void testFailToLaunchInterpreterProcess_ErrorInRunner() { try { System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_REMOTE_RUNNER.getVarName(), zeppelinHome.getAbsolutePath() + "/zeppelin-zengine/src/test/resources/bin/interpreter_invalid.sh"); final Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "sleep"); final InterpreterContext context1 = createDummyInterpreterContext(); try { interpreter1.interpret("1", context1); fail("Should not be able to launch interpreter process"); } catch (InterpreterException e) { assertTrue(ExceptionUtils.getStackTrace(e).contains("invalid_command:")); } } finally { System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_REMOTE_RUNNER.getVarName()); } }
@Test public void testFailToLaunchInterpreterProcess_Timeout() { try { System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_REMOTE_RUNNER.getVarName(), zeppelinHome.getAbsolutePath() + "/zeppelin-zengine/src/test/resources/bin/interpreter_timeout.sh"); System.setProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName(), "10000"); final Interpreter interpreter1 = interpreterSetting.getInterpreter("user1", "note1", "sleep"); final InterpreterContext context1 = createDummyInterpreterContext(); try { interpreter1.interpret("1", context1); fail("Should not be able to launch interpreter process"); } catch (InterpreterException e) { assertTrue(ExceptionUtils.getStackTrace(e).contains("Interpreter Process creation is time out")); } } finally { System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_REMOTE_RUNNER.getVarName()); System.clearProperty(ZeppelinConfiguration.ConfVars.ZEPPELIN_INTERPRETER_CONNECT_TIMEOUT.getVarName()); } }