This is an automated email from the ASF dual-hosted git repository.
zjffdu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/zeppelin.git
The following commit(s) were added to refs/heads/master by this push:
new a94ef00 [ZEPPELIN-4680]. Add method to release note memory and add
option to disable JobManager
a94ef00 is described below
commit a94ef005aaa7ec4f11ef752baa5737d2d837ae0d
Author: Jeff Zhang <zjf...@apache.org>
AuthorDate: Thu Mar 12 11:25:38 2020 +0800
[ZEPPELIN-4680]. Add method to release note memory and add option to
disable JobManager
### What is this PR for?
For zeppelin will load all notes in 2 places. One is
QuartzSchedulerService.java where it would read all notes and refresh their
cron job. Another is JobManagerService. In order to solve the memory issue, I
introduce the `unLoad` method in `Note` to release memory in
`QuartzSchedulerService` after refreshing cron job. Another approach is to
disable JobManager (Job tab in zeppelin UI), The Job tab is not so useful IMHO.
We should redesign that only to show the cron enabled jobs. But t [...]
### What type of PR is it?
[Improvement]
### Todos
* [ ] - Task
### What is the Jira issue?
* https://issues.apache.org/jira/browse/ZEPPELIN-4680
### How should this be tested?
* Manually tested
### 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 #3686 from zjffdu/ZEPPELIN-4680 and squashes the following commits:
a732f99e7 [Jeff Zhang] [ZEPPELIN-4680]. Add method to release note memory
and add option to disable JobManager
---
conf/zeppelin-site.xml.template | 7 +++
.../zeppelin/conf/ZeppelinConfiguration.java | 7 ++-
.../apache/zeppelin/service/JobManagerService.java | 14 +++++-
.../apache/zeppelin/cluster/ClusterEventTest.java | 3 +-
.../zeppelin/service/NotebookServiceTest.java | 3 +-
.../java/org/apache/zeppelin/notebook/Note.java | 13 +++++
.../org/apache/zeppelin/notebook/NoteManager.java | 24 +++++++++-
.../notebook/scheduler/NoSchedulerService.java | 4 +-
.../notebook/scheduler/QuartzSchedulerService.java | 55 +++++++++++++++++-----
.../notebook/scheduler/SchedulerService.java | 2 +-
.../interpreter/AbstractInterpreterTest.java | 1 +
.../org/apache/zeppelin/notebook/NotebookTest.java | 3 +-
12 files changed, 113 insertions(+), 23 deletions(-)
diff --git a/conf/zeppelin-site.xml.template b/conf/zeppelin-site.xml.template
index 9523be6..79e2f88 100755
--- a/conf/zeppelin-site.xml.template
+++ b/conf/zeppelin-site.xml.template
@@ -711,4 +711,11 @@
<description>path for storing search index on disk.</description>
</property>
+<property>
+ <name>zeppelin.jobmanager.enable</name>
+ <value>true</value>
+ <description>The Job tab in zeppelin page seems not so useful instead it
cost lots of memory and affect the performance.
+ Disable it can save lots of memory</description>
+</property>
+
</configuration>
diff --git
a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
index a57f133..15e004d 100644
---
a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
+++
b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/conf/ZeppelinConfiguration.java
@@ -609,6 +609,10 @@ public class ZeppelinConfiguration extends
XMLConfiguration {
return anonymousAllowed;
}
+ public boolean isJobManagerEnabled() {
+ return getBoolean(ConfVars.ZEPPELIN_JOBMANAGER_ENABLE);
+ }
+
public boolean isUsernameForceLowerCase() {
return getBoolean(ConfVars.ZEPPELIN_USERNAME_FORCE_LOWERCASE);
}
@@ -993,7 +997,8 @@ public class ZeppelinConfiguration extends XMLConfiguration
{
ZEPPELIN_PROXY_PASSWORD("zeppelin.proxy.password", null),
ZEPPELIN_SEARCH_INDEX_REBUILD("zeppelin.search.index.rebuild", false),
ZEPPELIN_SEARCH_USE_DISK("zeppelin.search.use.disk", true),
- ZEPPELIN_SEARCH_INDEX_PATH("zeppelin.search.index.path",
"/tmp/zeppelin-index");
+ ZEPPELIN_SEARCH_INDEX_PATH("zeppelin.search.index.path",
"/tmp/zeppelin-index"),
+ ZEPPELIN_JOBMANAGER_ENABLE("zeppelin.jobmanager.enable", true);
private String varName;
@SuppressWarnings("rawtypes")
diff --git
a/zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java
b/zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java
index 9798f03..f42534d 100644
---
a/zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java
+++
b/zeppelin-server/src/main/java/org/apache/zeppelin/service/JobManagerService.java
@@ -19,6 +19,7 @@ package org.apache.zeppelin.service;
import javax.inject.Inject;
import org.apache.commons.lang3.StringUtils;
+import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.Notebook;
import org.apache.zeppelin.notebook.Paragraph;
@@ -39,16 +40,21 @@ public class JobManagerService {
private static final Logger LOGGER =
LoggerFactory.getLogger(JobManagerService.class);
private Notebook notebook;
+ private ZeppelinConfiguration conf;
@Inject
- public JobManagerService(Notebook notebook) {
+ public JobManagerService(Notebook notebook, ZeppelinConfiguration conf) {
this.notebook = notebook;
+ this.conf = conf;
}
public List<NoteJobInfo> getNoteJobInfo(String noteId,
ServiceContext context,
ServiceCallback<List<NoteJobInfo>>
callback)
throws IOException {
+ if (!conf.isJobManagerEnabled()) {
+ return new ArrayList<>();
+ }
List<NoteJobInfo> notesJobInfo = new ArrayList<>();
Note jobNote = notebook.getNote(noteId);
if (jobNote == null) {
@@ -66,6 +72,9 @@ public class JobManagerService {
ServiceContext context,
ServiceCallback<List<NoteJobInfo>> callback)
throws IOException {
+ if (!conf.isJobManagerEnabled()) {
+ return new ArrayList<>();
+ }
List<Note> notes = notebook.getAllNotes();
List<NoteJobInfo> notesJobInfo = new ArrayList<>();
for (Note note : notes) {
@@ -81,6 +90,9 @@ public class JobManagerService {
public void removeNoteJobInfo(String noteId,
ServiceContext context,
ServiceCallback<List<NoteJobInfo>> callback)
throws IOException {
+ if (!conf.isJobManagerEnabled()) {
+ return;
+ }
List<NoteJobInfo> notesJobInfo = new ArrayList<>();
notesJobInfo.add(new NoteJobInfo(noteId, true));
callback.onSuccess(notesJobInfo, context);
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/cluster/ClusterEventTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/cluster/ClusterEventTest.java
index 912ed39..90ba134 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/cluster/ClusterEventTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/cluster/ClusterEventTest.java
@@ -86,7 +86,7 @@ public class ClusterEventTest extends ZeppelinServerMock {
private static Notebook notebook;
private static NotebookServer notebookServer;
- private static SchedulerService schedulerService;
+ private static QuartzSchedulerService schedulerService;
private static NotebookService notebookService;
private static AuthorizationService authorizationService;
private HttpServletRequest mockRequest;
@@ -103,6 +103,7 @@ public class ClusterEventTest extends ZeppelinServerMock {
authorizationService = TestUtils.getInstance(AuthorizationService.class);
schedulerService = new QuartzSchedulerService(zconf, notebook);
+ schedulerService.waitForFinishInit();
notebookServer = spy(NotebookServer.getInstance());
notebookService = new NotebookService(notebook, authorizationService,
zconf, schedulerService);
diff --git
a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
index 2a7de91..44a0e6f 100644
---
a/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
+++
b/zeppelin-server/src/test/java/org/apache/zeppelin/service/NotebookServiceTest.java
@@ -130,7 +130,8 @@ public class NotebookServiceTest {
credentials,
null);
- SchedulerService schedulerService = new
QuartzSchedulerService(zeppelinConfiguration, notebook);
+ QuartzSchedulerService schedulerService = new
QuartzSchedulerService(zeppelinConfiguration, notebook);
+ schedulerService.waitForFinishInit();
notebookService =
new NotebookService(
notebook, authorizationService, zeppelinConfiguration,
schedulerService);
diff --git
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
index 44fe45d..1c08fb1 100644
--- a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
+++ b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
@@ -180,6 +180,19 @@ public class Note implements JsonSerializable {
this.loaded = loaded;
}
+ /**
+ * Release note memory
+ */
+ public void unLoad() {
+ this.setLoaded(false);
+ this.paragraphs = null;
+ this.config = null;
+ this.info = null;
+ this.noteForms = null;
+ this.noteParams = null;
+ this.angularObjects = null;
+ }
+
public boolean isPersonalizedMode() {
Object v = getConfig().get("personalizedMode");
return null != v && "true".equals(v);
diff --git
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
index c260208..e19ece4 100644
---
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
+++
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/NoteManager.java
@@ -275,6 +275,22 @@ public class NoteManager {
* @return return null if not found on NotebookRepo.
* @throws IOException
*/
+ public Note getNote(String noteId, boolean forceLoad) throws IOException {
+ String notePath = this.notesInfo.get(noteId);
+ if (notePath == null) {
+ return null;
+ }
+ NoteNode noteNode = getNoteNode(notePath);
+ return noteNode.getNote(forceLoad);
+ }
+
+ /**
+ * Get note from NotebookRepo.
+ *
+ * @param noteId
+ * @return return null if not found on NotebookRepo.
+ * @throws IOException
+ */
public Note getNote(String noteId) throws IOException {
String notePath = this.notesInfo.get(noteId);
if (notePath == null) {
@@ -511,14 +527,18 @@ public class NoteManager {
this.notebookRepo = notebookRepo;
}
+ public synchronized Note getNote() throws IOException {
+ return getNote(true);
+ }
+
/**
* This method will load note from NotebookRepo. If you just want to get
noteId, noteName or
* notePath, you can call method getNoteId, getNoteName & getNotePath
* @return
* @throws IOException
*/
- public synchronized Note getNote() throws IOException {
- if (!note.isLoaded()) {
+ public synchronized Note getNote(boolean forceLoad) throws IOException {
+ if (!note.isLoaded() && forceLoad) {
note = notebookRepo.get(note.getId(), note.getPath(),
AuthenticationInfo.ANONYMOUS);
if (parent.toString().equals("/")) {
note.setPath("/" + note.getName());
diff --git
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/NoSchedulerService.java
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/NoSchedulerService.java
index 0263ec2..091b49e 100644
---
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/NoSchedulerService.java
+++
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/NoSchedulerService.java
@@ -22,8 +22,8 @@ import java.util.Set;
public class NoSchedulerService implements SchedulerService {
@Override
- public void refreshCron(String noteId) {
- // Do nothing
+ public boolean refreshCron(String noteId) {
+ return false;
}
@Override
diff --git
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/QuartzSchedulerService.java
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/QuartzSchedulerService.java
index cf1c0a3..416eddf 100644
---
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/QuartzSchedulerService.java
+++
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/QuartzSchedulerService.java
@@ -24,6 +24,7 @@ import java.util.Set;
import javax.inject.Inject;
+import com.google.common.annotations.VisibleForTesting;
import org.apache.zeppelin.conf.ZeppelinConfiguration;
import org.apache.zeppelin.notebook.Note;
import org.apache.zeppelin.notebook.Notebook;
@@ -47,6 +48,7 @@ public class QuartzSchedulerService implements
SchedulerService {
private final ZeppelinConfiguration zeppelinConfiguration;
private final Notebook notebook;
private final Scheduler scheduler;
+ private final Thread loadingNotesThread;
@Inject
public QuartzSchedulerService(ZeppelinConfiguration zeppelinConfiguration,
Notebook notebook)
@@ -58,10 +60,24 @@ public class QuartzSchedulerService implements
SchedulerService {
// Do in a separated thread because there may be many notes,
// loop all notes in the main thread may block the restarting of Zeppelin
server
- Thread loadingNotesThread = new Thread(() -> {
+ // TODO(zjffdu) It may cause issue when user delete note before this
thread is finished
+ this.loadingNotesThread = new Thread(() -> {
LOGGER.info("Starting init cronjobs");
notebook.getNotesInfo().stream()
- .forEach(entry -> refreshCron(entry.getId()));
+ .forEach(entry -> {
+ try {
+ if (!refreshCron(entry.getId())) {
+ try {
+ LOGGER.debug("Unload note: " + entry.getId());
+ notebook.getNote(entry.getId()).unLoad();
+ } catch (Exception e) {
+ LOGGER.warn("Fail to unload note: " + entry.getId(),
e);
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.warn("Fail to refresh cron for note: " +
entry.getId());
+ }
+ });
LOGGER.info("Complete init cronjobs");
});
loadingNotesThread.setName("Init CronJob Thread");
@@ -69,40 +85,52 @@ public class QuartzSchedulerService implements
SchedulerService {
loadingNotesThread.start();
}
+ /**
+ * This is only for testing, unit test should always call this method in
setup() before testing.
+ */
+ @VisibleForTesting
+ public void waitForFinishInit() {
+ try {
+ loadingNotesThread.join();
+ } catch (InterruptedException e) {
+ LOGGER.warn("Unexpected exception", e);
+ }
+ }
+
@Override
- public void refreshCron(String noteId) {
+ public boolean refreshCron(String noteId) {
removeCron(noteId);
Note note = null;
try {
note = notebook.getNote(noteId);
} catch (IOException e) {
LOGGER.warn("Skip refresh cron of note: " + noteId + " because fail to
get it", e);
- return;
+ return false;
}
if (note == null) {
LOGGER.warn("Skip refresh cron of note: " + noteId + " because there's
no such note");
- return;
+ return false;
}
if (note.isTrash()) {
LOGGER.warn("Skip refresh cron of note: " + noteId + " because it is in
trash");
- return;
+ return false;
}
Map<String, Object> config = note.getConfig();
if (config == null) {
LOGGER.warn("Skip refresh cron of note: " + noteId + " because its
config is empty.");
- return;
+ return false;
}
if (!note.isCronSupported(zeppelinConfiguration)) {
LOGGER.warn("Skip refresh cron of note " + noteId + " because its cron
is not enabled.");
- return;
+ return false;
}
String cronExpr = (String) note.getConfig().get("cron");
if (cronExpr == null || cronExpr.trim().length() == 0) {
LOGGER.warn("Skip refresh cron of note " + noteId + " because its cron
expression is empty.");
- return;
+ return false;
}
JobDataMap jobDataMap =
@@ -132,16 +160,17 @@ public class QuartzSchedulerService implements
SchedulerService {
} catch (Exception e) {
LOGGER.error("Fail to create cron trigger for note: " + note.getName(),
e);
info.put("cron", e.getMessage());
+ return false;
}
try {
- if (trigger != null) {
- LOGGER.info("Trigger cron for note: " + note.getName() + ", with cron
expression: " + cronExpr);
- scheduler.scheduleJob(newJob, trigger);
- }
+ LOGGER.info("Trigger cron for note: " + note.getName() + ", with cron
expression: " + cronExpr);
+ scheduler.scheduleJob(newJob, trigger);
+ return true;
} catch (SchedulerException e) {
LOGGER.error("Fail to schedule cron job for note: " + note.getName(), e);
info.put("cron", "Scheduler Exception");
+ return false;
}
}
diff --git
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/SchedulerService.java
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/SchedulerService.java
index b7c87af..045b694 100644
---
a/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/SchedulerService.java
+++
b/zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/scheduler/SchedulerService.java
@@ -20,6 +20,6 @@ package org.apache.zeppelin.notebook.scheduler;
import java.util.Set;
public interface SchedulerService {
- void refreshCron(String noteId);
+ boolean refreshCron(String noteId);
Set<?> getJobs();
}
diff --git
a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
index 4f975fb..d4297bf 100644
---
a/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
+++
b/zeppelin-zengine/src/test/java/org/apache/zeppelin/interpreter/AbstractInterpreterTest.java
@@ -42,6 +42,7 @@ public abstract class AbstractInterpreterTest {
interpreterDir = new File(zeppelinHome, "interpreter_" +
getClass().getSimpleName());
confDir = new File(zeppelinHome, "conf_" + getClass().getSimpleName());
notebookDir = new File(zeppelinHome, "notebook_" +
getClass().getSimpleName());
+ FileUtils.deleteDirectory(notebookDir);
interpreterDir.mkdirs();
confDir.mkdirs();
diff --git
a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
index 529c96e..e35079c 100644
---
a/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
+++
b/zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java
@@ -83,7 +83,7 @@ public class NotebookTest extends AbstractInterpreterTest
implements ParagraphJo
private Credentials credentials;
private AuthenticationInfo anonymous = AuthenticationInfo.ANONYMOUS;
private StatusChangedListener afterStatusChangedListener;
- private SchedulerService schedulerService;
+ private QuartzSchedulerService schedulerService;
@Before
public void setUp() throws Exception {
@@ -102,6 +102,7 @@ public class NotebookTest extends AbstractInterpreterTest
implements ParagraphJo
credentials, null);
notebook.setParagraphJobListener(this);
schedulerService = new QuartzSchedulerService(conf, notebook);
+ schedulerService.waitForFinishInit();
}
@After
