Elgato_dark/zeppelin
1
0
Fork 0
mirror of https://github.com/apache/zeppelin synced 2026-05-24 09:38:26 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fix scoped mode and add unit test

Browse source
This commit is contained in:
Jeff Zhang 2017-01-10 14:07:33 +08:00
parent 8cb28a31b9
commit 5ee076dd93
2 changed files with 97 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
View file

@ -144,6 +144,26 @@ public class InterpreterSetting {
return key;
}
private String getInterpreterSessionKey(String user, String noteId) {
InterpreterOption option = getOption();
String key;
if (option.isExistingProcess()) {
key = Constants.EXISTING_PROCESS;
} else if (option.perNoteScoped() && option.perUserScoped()) {
key = user + ":" + noteId;
} else if (option.perUserScoped()) {
key = user;
} else if (option.perNoteScoped()) {
key = noteId;
} else {
key = "shared_session";
}
logger.debug("Interpreter session key: {}, for note: {}, user: {}, InterpreterSetting Name: " +
"{}", key, noteId, user, getName());
return key;
}
public InterpreterGroup getInterpreterGroup(String user, String noteId) {
String key = getInterpreterProcessKey(user, noteId);
if (!interpreterGroupRef.containsKey(key)) {
@ -194,11 +214,11 @@ public class InterpreterSetting {
if (user.equals("anonymous")) {
user = "";
}
String key = getInterpreterProcessKey(user, "");
String processKey = getInterpreterProcessKey(user, "");
String sessionKey = getInterpreterSessionKey(user, "");
InterpreterGroup groupToRemove = null;
for (String intpKey : new HashSet<>(interpreterGroupRef.keySet())) {
if (intpKey.contains(key)) {
if (intpKey.contains(processKey)) {
interpreterGroupWriteLock.lock();
groupToRemove = interpreterGroupRef.remove(intpKey);
interpreterGroupWriteLock.unlock();
@ -206,7 +226,7 @@ public class InterpreterSetting {
}
if (groupToRemove != null) {
groupToRemove.close();
groupToRemove.close(sessionKey);
}
}

78
zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterFactoryTest.java
View file

@ -167,6 +167,79 @@ public class InterpreterFactoryTest {
assertEquals("value_2", remoteInterpreter.getProperty("property_2"));
}
/**
* 2 users' interpreters in scoped mode. Each user has one session. Restarting user1's interpreter
* won't affect user2's interpreter
* @throws Exception
*/
@Test
public void testRestartInterpreterInScopedMode() throws Exception {
factory = new InterpreterFactory(conf, new InterpreterOption(true), null, null, null, depResolver, false);
List<InterpreterSetting> all = factory.get();
InterpreterSetting mock1Setting = null;
for (InterpreterSetting setting : all) {
if (setting.getName().equals("mock1")) {
mock1Setting = setting;
break;
}
}
mock1Setting.getOption().setPerUser("scoped");
mock1Setting.getOption().setPerNote("shared");
// set remote as false so that we won't create new remote interpreter process
mock1Setting.getOption().setRemote(false);
mock1Setting.getOption().setHost("localhost");
mock1Setting.getOption().setPort(2222);
InterpreterGroup interpreterGroup = mock1Setting.getInterpreterGroup("user1", "sharedProcess");
factory.createInterpretersForNote(mock1Setting, "user1", "sharedProcess", "user1");
factory.createInterpretersForNote(mock1Setting, "user2", "sharedProcess", "user2");
LazyOpenInterpreter interpreter1 = (LazyOpenInterpreter)interpreterGroup.get("user1").get(0);
interpreter1.open();
LazyOpenInterpreter interpreter2 = (LazyOpenInterpreter)interpreterGroup.get("user2").get(0);
interpreter2.open();
mock1Setting.closeAndRemoveInterpreterGroupByUser("user1");
assertFalse(interpreter1.isOpen());
assertTrue(interpreter2.isOpen());
}
/**
* 2 users' interpreters in isolated mode. Each user has one interpreterGroup. Restarting user1's interpreter
* won't affect user2's interpreter
* @throws Exception
*/
@Test
public void testRestartInterpreterInIsolatedMode() throws Exception {
factory = new InterpreterFactory(conf, new InterpreterOption(true), null, null, null, depResolver, false);
List<InterpreterSetting> all = factory.get();
InterpreterSetting mock1Setting = null;
for (InterpreterSetting setting : all) {
if (setting.getName().equals("mock1")) {
mock1Setting = setting;
break;
}
}
mock1Setting.getOption().setPerUser("isolated");
mock1Setting.getOption().setPerNote("shared");
// set remote as false so that we won't create new remote interpreter process
mock1Setting.getOption().setRemote(false);
mock1Setting.getOption().setHost("localhost");
mock1Setting.getOption().setPort(2222);
InterpreterGroup interpreterGroup1 = mock1Setting.getInterpreterGroup("user1", "note1");
InterpreterGroup interpreterGroup2 = mock1Setting.getInterpreterGroup("user2", "note2");
factory.createInterpretersForNote(mock1Setting, "user1", "note1", "shared_session");
factory.createInterpretersForNote(mock1Setting, "user2", "note2", "shared_session");
LazyOpenInterpreter interpreter1 = (LazyOpenInterpreter)interpreterGroup1.get("shared_session").get(0);
interpreter1.open();
LazyOpenInterpreter interpreter2 = (LazyOpenInterpreter)interpreterGroup2.get("shared_session").get(0);
interpreter2.open();
mock1Setting.closeAndRemoveInterpreterGroupByUser("user1");
assertFalse(interpreter1.isOpen());
assertTrue(interpreter2.isOpen());
}
@Test
public void testFactoryDefaultList() throws IOException, RepositoryException {
// get default settings
@ -365,9 +438,4 @@ public class InterpreterFactoryTest {
interpreterRunner = ((RemoteInterpreter) ((LazyOpenInterpreter) i).getInnerInterpreter()).getInterpreterRunner();
assertEquals(interpreterRunner, testInterpreterRunner);
}
@Test
public void interpreterRunnerAsAbsolutePathTest() {
}
}