zeppelin git commit: [ZEPPELIN-525] Test failing in zeppelin-interpreter

[fe...@apache.org]

Repository: zeppelin
Updated Branches:
  refs/heads/master e763b3bf3 -> 65e1d3645


[ZEPPELIN-525] Test failing in zeppelin-interpreter

`RemoteSchedulerTest.test` was fall when CPU switched `RemoteScheduler` execution (which executes a job in a separate thread) back to test main thread execution after changing job status to `Status.FINISHED` but before updating running jobs list:
```
job.setStatus(lastStatus);

// if thread execution is switched here to the main thread test will fall

synchronized (queue) {
    running.remove(job);
    queue.notify();
}
```
Test checked job status, saw that job was terminated and expected to see an empty jobs runing list.
```
while (!job.isTerminated() && cycles < MAX_WAIT_CYCLES) {
    Thread.sleep(TICK_WAIT);
    cycles++;
}

// this assert will fail because a job was not removed from running list yet
assertEquals(0, scheduler.getJobsRunning().size());
```
But `scheduler.running` list still contained the last job and the assertion failed.

### What is this PR for?
The goal is to synchronize updating of a job status and a scheduler running jobs list when a job is terminated.

### What type of PR is it?
Bug Fix

### What is the Jira issue?
[ZEPPELIN-525](https://issues.apache.org/jira/browse/ZEPPELIN-525)

### How should this be tested?
You may place a `Thread.sleep(1000);` code to the `RemoteScheduler.java` class after the `job.setStatus(lastStatus);` (line number 354). This will increase probability of threads switching after this line. Test should not fail.

### Questions:
* Does the licenses files need update? **no**
* Is there breaking changes for older versions? **no**
* Does this needs documentation? **no**

Author: Alexander Shoshin <Al...@epam.com>

Closes #1954 from AlexanderShoshin/ZEPPELIN-525 and squashes the following commits:

8a0f4af [Alexander Shoshin] synchronized a job status and a list of running jobs


Project: http://git-wip-us.apache.org/repos/asf/zeppelin/repo
Commit: http://git-wip-us.apache.org/repos/asf/zeppelin/commit/65e1d364
Tree: http://git-wip-us.apache.org/repos/asf/zeppelin/tree/65e1d364
Diff: http://git-wip-us.apache.org/repos/asf/zeppelin/diff/65e1d364

Branch: refs/heads/master
Commit: 65e1d364583ac8fb822aa00ebee5b7cfcd842fed
Parents: e763b3b
Author: Alexander Shoshin <Al...@epam.com>
Authored: Fri Jan 27 11:30:11 2017 +0300
Committer: Felix Cheung <fe...@apache.org>
Committed: Sun Jan 29 23:35:53 2017 -0800

----------------------------------------------------------------------
 .../zeppelin/scheduler/RemoteScheduler.java     | 20 ++++++++++----------
 1 file changed, 10 insertions(+), 10 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/zeppelin/blob/65e1d364/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
----------------------------------------------------------------------
diff --git a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
index 0101b18..a4ab00e 100644
--- a/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
+++ b/zeppelin-interpreter/src/main/java/org/apache/zeppelin/scheduler/RemoteScheduler.java
@@ -307,10 +307,10 @@ public class RemoteScheduler implements Scheduler {
     @Override
     public void run() {
       if (job.isAborted()) {
-        job.setStatus(Status.ABORT);
-        job.aborted = false;
-
         synchronized (queue) {
+          job.setStatus(Status.ABORT);
+          job.aborted = false;
+
           running.remove(job);
           queue.notify();
         }
@@ -350,16 +350,16 @@ public class RemoteScheduler implements Scheduler {
         lastStatus = Status.ERROR;
       }
 
-      job.setStatus(lastStatus);
+      synchronized (queue) {
+        job.setStatus(lastStatus);
 
-      if (listener != null) {
-        listener.jobFinished(scheduler, job);
-      }
+        if (listener != null) {
+          listener.jobFinished(scheduler, job);
+        }
 
-      // reset aborted flag to allow retry
-      job.aborted = false;
+        // reset aborted flag to allow retry
+        job.aborted = false;
 
-      synchronized (queue) {
         running.remove(job);
         queue.notify();
       }