[ZEPPELIN-3077] Cron scheduler is easy to get stuck when one of the cron jobs takes long time or gets stuck

docs/_includes/themes/zeppelin/_navigation.html

@@ -62,6 +62,7 @@
                 <li><a href="{{BASE_PATH}}/usage/other_features/personalized_mode.html">Personalized Mode</a></li>
                 <li><a href="{{BASE_PATH}}/usage/other_features/customizing_homepage.html">Customizing Zeppelin Homepage</a></li>
                 <li><a href="{{BASE_PATH}}/usage/other_features/notebook_actions.html">Notebook Actions</a></li>
+                <li><a href="{{BASE_PATH}}/usage/other_features/cron_scheduler.html">Cron Scheduler</a></li>
                 <li role="separator" class="divider"></li>
                 <li class="title"><span>REST API</span></li>
                 <li><a href="{{BASE_PATH}}/usage/rest_api/interpreter.html">Interpreter API</a></li>

docs/usage/other_features/cron_scheduler.md

@@ -0,0 +1,52 @@
+---
+layout: page
+title: "Running a Notebook on a Given Schedule Automatically"
+description: "You can run a notebook on a given schedule automatically by setting up a cron scheduler on the notebook."
+group: usage/other_features
+---
+<!--
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+-->
+{% include JB/setup %}
+
+# Running a Notebook on a Given Schedule Automatically
+
+<div id="toc"></div>
+
+Apache Zeppelin provides a cron scheduler for each notebook. You can run a notebook on a given schedule automatically by setting up a cron scheduler on the notebook.
+
+## Setting up a cron scheduler on a notebook
+
+Click the clock icon on the tool bar and open a cron scheduler dialog box.
+
+<img src="{{BASE_PATH}}/assets/themes/zeppelin/img/docs-img/cron_scheduler_dialog_box.png" />
+
+There are the following items which you can input or set:
+
+### Preset
+
+You can set a cron schedule easily by clicking each option such as `1m` and `5m`. The login user is set as a cron executing user automatically. You can also clear the cron schedule settings by clicking `None`.
+
+### Cron expression
+
+You can set the cron schedule by filling in this form. Please see [Cron Trigger Tutorial](http://www.quartz-scheduler.org/documentation/quartz-2.2.x/tutorials/crontrigger) for the available cron syntax.
+
+### Cron executing user
+
+You can set the cron executing user by filling in this form and press the enter key.
+
+### auto-restart interpreter on cron execution
+
+When this checkbox is set to "on", the interpreters which are binded to the notebook are stopped automatically after the cron execution. This feature is useful if you want to release the interpreter resources after the cron execution.
+
+> **Note**: A cron execution is skipped if one of the paragraphs is in a state of `RUNNING` or `PENDING` no matter whether it is executed automatically (i.e. by the cron scheduler) or manually by a user opening this notebook.

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java

@@ -652,6 +652,22 @@ public class Note implements ParagraphJobListener, JsonSerializable {
     return true;
   }
 
+  /**
+   * Return true if there is a running or pending paragraph
+   */
+  boolean isRunningOrPending() {
+    synchronized (paragraphs) {
+      for (Paragraph p : paragraphs) {
+        Status status = p.getStatus();
+        if (status.isRunning() || status.isPending()) {
+          return true;
+        }
+      }
+    }
+
+    return false;
+  }
+
   public boolean isTrash() {
     String path = getName();
     if (path.charAt(0) == '/') {

zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Notebook.java

@@ -890,16 +890,15 @@ public class Notebook implements NoteEventListener {
 
       String noteId = context.getJobDetail().getJobDataMap().getString("noteId");
       Note note = notebook.getNote(noteId);
-      note.runAll();
 
-      while (!note.isTerminated()) {
-        try {
-          Thread.sleep(1000);
-        } catch (InterruptedException e) {
-          logger.error(e.toString(), e);
-        }
+      if (note.isRunningOrPending()) {
+        logger.warn("execution of the cron job is skipped because there is a running or pending " +
+            "paragraph (note id: {})", noteId);
+        return;
       }
 
+      note.runAll();
+
       boolean releaseResource = false;
       String cronExecutingUser = null;
       try {

zeppelin-zengine/src/test/java/org/apache/zeppelin/notebook/NotebookTest.java

@@ -361,6 +361,46 @@ public class NotebookTest extends AbstractInterpreterTest implements JobListener
     notebook.removeNote(note.getId(), anonymous);
   }
 
+  @Test
+  public void testScheduleAgainstRunningAndPendingParagraph() throws InterruptedException, IOException {
+    // create a note
+    Note note = notebook.createNote(anonymous);
+    interpreterSettingManager.setInterpreterBinding("user", note.getId(),
+            interpreterSettingManager.getInterpreterSettingIds());
+
+    // append running and pending paragraphs to the note
+    for (Status status: new Status[]{Status.RUNNING, Status.PENDING}) {
+      Paragraph p = note.addNewParagraph(AuthenticationInfo.ANONYMOUS);
+      Map config = new HashMap<>();
+      p.setConfig(config);
+      p.setText("p");
+      p.setStatus(status);
+      assertNull(p.getDateFinished());
+    }
+
+    // set cron scheduler, once a second
+    Map config = note.getConfig();
+    config.put("enabled", true);
+    config.put("cron", "* * * * * ?");
+    note.setConfig(config);
+    notebook.refreshCron(note.getId());
+    Thread.sleep(2 * 1000);
+
+    // remove cron scheduler.
+    config.put("cron", null);
+    note.setConfig(config);
+    notebook.refreshCron(note.getId());
+    Thread.sleep(2 * 1000);
+
+    // check if the executions of the running and pending paragraphs were skipped
+    for (Paragraph p : note.paragraphs) {
+      assertNull(p.getDateFinished());
+    }
+
+    // remove the note
+    notebook.removeNote(note.getId(), anonymous);
+  }
+
   @Test
   public void testSchedulePoolUsage() throws InterruptedException, IOException {
     final int timeout = 30;