Skip to content

ZEPPELIN-2998. Fix bug in restarting interpreter in scoped mode #2626

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zjffdu wants to merge 1 commit into from
Closed

ZEPPELIN-2998. Fix bug in restarting interpreter in scoped mode #2626

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 6 additions & 7 deletions zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSetting.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,10 +205,10 @@ public Builder setDependencyResolver(DependencyResolver dependencyResolver) {
return this;
}

// public Builder setInterpreterRunner(InterpreterRunner runner) {
// interpreterSetting.interpreterRunner = runner;
// return this;
// }
public Builder setInterpreterRunner(InterpreterRunner runner) {
interpreterSetting.interpreterRunner = runner;
return this;
}

public Builder setIntepreterSettingManager(
InterpreterSettingManager interpreterSettingManager) {
Expand Down Expand Up @@ -248,7 +248,6 @@ public InterpreterSetting() {
}

void postProcessing() {
// createLauncher();
this.status = Status.READY;
}

Expand Down Expand Up @@ -370,7 +369,7 @@ public ManagedInterpreterGroup getOrCreateInterpreterGroup(String user, String n
try {
interpreterGroupWriteLock.lock();
if (!interpreterGroups.containsKey(groupId)) {
LOGGER.info("Create InterpreterGroup with groupId {} for user {} and note {}",
LOGGER.info("Create InterpreterGroup with groupId: {} for user: {} and note: {}",
groupId, user, noteId);
ManagedInterpreterGroup intpGroup = createInterpreterGroup(groupId);
interpreterGroups.put(groupId, intpGroup);
Expand Down Expand Up @@ -653,7 +652,7 @@ synchronized RemoteInterpreterProcess createInterpreterProcess() throws IOExcept
return process;
}

private List<Interpreter> getOrCreateSession(String user, String noteId) {
List<Interpreter> getOrCreateSession(String user, String noteId) {
ManagedInterpreterGroup interpreterGroup = getOrCreateInterpreterGroup(user, noteId);
Preconditions.checkNotNull(interpreterGroup, "No InterpreterGroup existed for user {}, " +
"noteId {}", user, noteId);
Expand Down
8 changes: 1 addition & 7 deletions ...elin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterSettingManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -782,13 +782,7 @@ public void restart(String settingId, String noteId, String user) throws Interpr
//clean up metaInfos
intpSetting.setInfos(null);
copyDependenciesFromLocalPath(intpSetting);

if (user.equals("anonymous")) {
intpSetting.close();
} else {
intpSetting.closeInterpreters(user, noteId);
}

intpSetting.closeInterpreters(user, noteId);
} else {
throw new InterpreterException("Interpreter setting id " + settingId + " not found");
}
Expand Down
4 changes: 2 additions & 2 deletions zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/ManagedInterpreterGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ public synchronized void close(String sessionId) {
close(sessions.remove(sessionId));
//TODO(zjffdu) whether close InterpreterGroup if there's no session left in Zeppelin Server
if (sessions.isEmpty() && interpreterSetting != null) {
LOGGER.info("Remove this InterpreterGroup {} as all the sessions are closed", id);
LOGGER.info("Remove this InterpreterGroup: {} as all the sessions are closed", id);
interpreterSetting.removeInterpreterGroup(id);
if (remoteInterpreterProcess != null) {
LOGGER.info("Kill RemoteIntetrpreterProcess");
Expand Down Expand Up @@ -133,7 +133,7 @@ public synchronized List<Interpreter> getOrCreateSession(String user, String ses
for (Interpreter interpreter : interpreters) {
interpreter.setInterpreterGroup(this);
}
LOGGER.info("Create Session {} in InterpreterGroup {} for user {}", sessionId, id, user);
LOGGER.info("Create Session: {} in InterpreterGroup: {} for user: {}", sessionId, id, user);
sessions.put(sessionId, interpreters);
return interpreters;
}
Expand Down
77 changes: 74 additions & 3 deletions ...-zengine/src/test/java/org/apache/zeppelin/interpreter/InterpreterSettingManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import java.util.Map;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
Expand Down Expand Up @@ -181,9 +182,6 @@ public void testCreateUpdateRemoveSetting() throws IOException, InterpreterExcep
// only close user1's session
interpreterSettingManager.restart(interpreterSetting.getId(), "note1", "user1");
assertEquals(2, interpreterGroup.getSessionNum());
// close all the sessions
interpreterSettingManager.restart(interpreterSetting.getId(), "note1", "anonymous");
assertEquals(0, interpreterGroup.getSessionNum());

// remove interpreter setting
interpreterSettingManager.remove(interpreterSetting.getId());
Expand Down Expand Up @@ -281,6 +279,79 @@ public void testGetEditor() throws IOException {
Interpreter mock1Interpreter = interpreterFactory.getInterpreter("user1", "note1", "mock1");
editor = interpreterSettingManager.getEditorSetting(mock1Interpreter,"user1", "note1", "mock1");
assertEquals("text", editor.get("language"));
}

@Test
public void testRestartShared() throws InterpreterException {
InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
interpreterSetting.getOption().setPerUser("shared");
interpreterSetting.getOption().setPerNote("shared");

interpreterSetting.getOrCreateSession("user1", "note1");
interpreterSetting.getOrCreateInterpreterGroup("user2", "note2");
assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());

interpreterSettingManager.restart(interpreterSetting.getId(), "user1", "note1");
assertEquals(0, interpreterSetting.getAllInterpreterGroups().size());
}

@Test
public void testRestartPerUserIsolated() throws InterpreterException {
InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
interpreterSetting.getOption().setPerUser("isolated");
interpreterSetting.getOption().setPerNote("shared");

interpreterSetting.getOrCreateSession("user1", "note1");
interpreterSetting.getOrCreateSession("user2", "note2");
assertEquals(2, interpreterSetting.getAllInterpreterGroups().size());

interpreterSettingManager.restart(interpreterSetting.getId(), "note1", "user1");
assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
}

@Test
public void testRestartPerNoteIsolated() throws InterpreterException {
InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
interpreterSetting.getOption().setPerUser("shared");
interpreterSetting.getOption().setPerNote("isolated");

interpreterSetting.getOrCreateSession("user1", "note1");
interpreterSetting.getOrCreateSession("user2", "note2");
assertEquals(2, interpreterSetting.getAllInterpreterGroups().size());

interpreterSettingManager.restart(interpreterSetting.getId(), "note1", "user1");
assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
}

@Test
public void testRestartPerUserScoped() throws InterpreterException {
InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
interpreterSetting.getOption().setPerUser("scoped");
interpreterSetting.getOption().setPerNote("shared");

interpreterSetting.getOrCreateSession("user1", "note1");
interpreterSetting.getOrCreateSession("user2", "note2");
assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
assertEquals(2, interpreterSetting.getAllInterpreterGroups().get(0).getSessionNum());

interpreterSettingManager.restart(interpreterSetting.getId(), "note1", "user1");
assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
assertEquals(1, interpreterSetting.getAllInterpreterGroups().get(0).getSessionNum());
}

@Test
public void testRestartPerNoteScoped() throws InterpreterException {
InterpreterSetting interpreterSetting = interpreterSettingManager.getByName("test");
interpreterSetting.getOption().setPerUser("shared");
interpreterSetting.getOption().setPerNote("scoped");

interpreterSetting.getOrCreateSession("user1", "note1");
interpreterSetting.getOrCreateSession("user2", "note2");
assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
assertEquals(2, interpreterSetting.getAllInterpreterGroups().get(0).getSessionNum());

interpreterSettingManager.restart(interpreterSetting.getId(), "note1", "user1");
assertEquals(1, interpreterSetting.getAllInterpreterGroups().size());
assertEquals(1, interpreterSetting.getAllInterpreterGroups().get(0).getSessionNum());
}
}