Fix the regression issue of get Spark job done on Spark 2.2

Signed-off-by: Wei Zhang <wezhang@outlook.com>
microsoft · Apr 25, 2018 · 596878d · 596878d
1 parent 71c5969
commit 596878d
Showing 1 changed file with 7 additions and 4 deletions.
diff --git a/...s/hdinsight-node-common/src/com/microsoft/azure/hdinsight/spark/common/SparkBatchJob.java b/...s/hdinsight-node-common/src/com/microsoft/azure/hdinsight/spark/common/SparkBatchJob.java
@@ -750,11 +750,11 @@ public boolean isActive() throws IOException {
     public Observable<SimpleImmutableEntry<SparkBatchJobState, String>> getJobDoneObservable() {
         return Observable.create((Subscriber<? super SimpleImmutableEntry<SparkBatchJobState, String>> ob) -> {
             try {
-                boolean isJobActive = true;
+                boolean isJobActive;
                 SparkBatchJobState state = SparkBatchJobState.NOT_STARTED;
                 String diagnostics = "";
 
-                while (isJobActive) {
+                do {
                     HttpResponse httpResponse = this.getSubmission().getBatchSparkJobStatus(
                             this.getConnectUri().toString(), batchId);
 
@@ -768,17 +768,20 @@ public Observable<SimpleImmutableEntry<SparkBatchJobState, String>> getJobDoneOb
                         diagnostics = String.join("\n", jobResp.getLog());
 
                         isJobActive = !state.isJobDone();
+                    } else {
+                        isJobActive = false;
                     }
 
+
                     // Retry interval
                     sleep(1000);
-                }
+                } while (isJobActive);
 
                 ob.onNext(new SimpleImmutableEntry<>(state, diagnostics));
+                ob.onCompleted();
             } catch (IOException ex) {
                 ob.onError(ex);
             } catch (InterruptedException ignored) {
-            } finally {
                 ob.onCompleted();
             }
         });