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

ZEPPELIN-1770. Restart only the client user's interpreter when restarting interpreter setting

Browse source
This commit is contained in:
Jeff Zhang 2016-12-14 15:04:20 +08:00
parent 24ca17b7df
commit 8cb28a31b9
3 changed files with 43 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
zeppelin-server/src/main/java/org/apache/zeppelin/rest/InterpreterRestApi.java
View file

@ -38,6 +38,7 @@ import javax.ws.rs.core.Response.Status;
import com.google.gson.Gson;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.zeppelin.rest.message.RestartInterpreterRequest;
import org.apache.zeppelin.utils.SecurityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sonatype.aether.repository.RemoteRepository;
@ -178,12 +179,11 @@ public class InterpreterRestApi {
@ZeppelinApi
public Response restartSetting(String message, @PathParam("settingId") String settingId) {
logger.info("Restart interpreterSetting {}, msg={}", settingId, message);
try {
RestartInterpreterRequest request = gson.fromJson(message, RestartInterpreterRequest.class);
String noteId = request == null ? null : request.getNoteId();
interpreterFactory.restart(settingId, noteId);
interpreterFactory.restart(settingId, noteId, SecurityUtils.getPrincipal());
} catch (InterpreterException e) {
logger.error("Exception in InterpreterRestApi while restartSetting ", e);

26
zeppelin-zengine/src/main/java/org/apache/zeppelin/interpreter/InterpreterFactory.java
View file

@ -739,7 +739,7 @@ public class InterpreterFactory implements InterpreterGroupFactory {
String noteId) {
InterpreterOption option = interpreterSetting.getOption();
if (option.isProcess()) {
interpreterSetting.closeAndRemoveInterpreterGroup(noteId);
interpreterSetting.closeAndRemoveInterpreterGroupByNoteId(noteId);
} else if (option.isSession()) {
InterpreterGroup interpreterGroup = interpreterSetting.getInterpreterGroup(user, noteId);
String key = getInterpreterSessionKey(user, noteId, interpreterSetting);
@ -973,18 +973,23 @@ public class InterpreterFactory implements InterpreterGroupFactory {
return noteId == null ? false : true;
}
public void restart(String settingId, String noteId) {
public void restart(String settingId, String noteId, String user) {
InterpreterSetting intpSetting = interpreterSettings.get(settingId);
Preconditions.checkNotNull(intpSetting);
// restart interpreter setting in note page
if (noteIdIsExist(noteId) && intpSetting.getOption().isProcess()) {
intpSetting.closeAndRemoveInterpreterGroup(noteId);
intpSetting.closeAndRemoveInterpreterGroupByNoteId(noteId);
return;
} else {
// restart interpreter setting in interpreter setting page
restart(settingId, user);
}
restart(settingId);
}
public void restart(String id) {
public void restart(String id, String user) {
synchronized (interpreterSettings) {
InterpreterSetting intpSetting = interpreterSettings.get(id);
// Check if dependency in specified path is changed
@ -995,8 +1000,11 @@ public class InterpreterFactory implements InterpreterGroupFactory {
copyDependenciesFromLocalPath(intpSetting);
stopJobAllInterpreter(intpSetting);
intpSetting.closeAndRemoveAllInterpreterGroups();
if (user.equals("anonymous")) {
intpSetting.closeAndRemoveAllInterpreterGroups();
} else {
intpSetting.closeAndRemoveInterpreterGroupByUser(user);
}
} else {
throw new InterpreterException("Interpreter setting id " + id + " not found");
@ -1004,6 +1012,10 @@ public class InterpreterFactory implements InterpreterGroupFactory {
}
}
public void restart(String id) {
restart(id, "anonymous");
}
private void stopJobAllInterpreter(InterpreterSetting intpSetting) {
if (intpSetting != null) {
for (InterpreterGroup intpGroup : intpSetting.getAllInterpreterGroups()) {

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

@ -173,7 +173,7 @@ public class InterpreterSetting {
}
}
void closeAndRemoveInterpreterGroup(String noteId) {
void closeAndRemoveInterpreterGroupByNoteId(String noteId) {
String key = getInterpreterProcessKey("", noteId);
InterpreterGroup groupToRemove = null;
@ -190,10 +190,30 @@ public class InterpreterSetting {
}
}
void closeAndRemoveInterpreterGroupByUser(String user) {
if (user.equals("anonymous")) {
user = "";
}
String key = getInterpreterProcessKey(user, "");
InterpreterGroup groupToRemove = null;
for (String intpKey : new HashSet<>(interpreterGroupRef.keySet())) {
if (intpKey.contains(key)) {
interpreterGroupWriteLock.lock();
groupToRemove = interpreterGroupRef.remove(intpKey);
interpreterGroupWriteLock.unlock();
}
}
if (groupToRemove != null) {
groupToRemove.close();
}
}
void closeAndRemoveAllInterpreterGroups() {
HashSet<String> groupsToRemove = new HashSet<>(interpreterGroupRef.keySet());
for (String key : groupsToRemove) {
closeAndRemoveInterpreterGroup(key);
closeAndRemoveInterpreterGroupByNoteId(key);
}
}