Fix the job done observable complete issue

Signed-off-by: Wei Zhang <[email protected]>

Author: wezhang

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/azure/hdinsight/spark/run/SparkBatchJobRemoteProcess.java

@@ -187,6 +187,7 @@ public void start() {
                                 sparkJob = job;
 
                                 ob.onNext(job);
+                                ob.onCompleted();
                             } catch (IOException e) {
                                 ob.onError(e);
                             }
@@ -213,10 +214,11 @@ public void start() {
                         ctrlSubject.onNext(new SimpleImmutableEntry<>(MessageInfoType.Error, "Job state is " + sdPair.getKey().toString()));
                         ctrlSubject.onNext(new SimpleImmutableEntry<>(MessageInfoType.Error, "Diagnostics: " + sdPair.getValue()));
                     }
-                    disconnect();
                 }, err -> {
                     ctrlSubject.onError(err);
                     disconnect();
+                }, () -> {
+                    disconnect();
                 });
     }
 

Utils/hdinsight-node-common/src/com/microsoft/azure/hdinsight/spark/common/SparkBatchJob.java

@@ -517,7 +517,7 @@ public Observable<SimpleImmutableEntry<SparkBatchJobState, String>> getJobDoneOb
                 String applicationId = null;
                 String diagnostics = "";
 
-                while (true) {
+                while (isJobActive || !isLogAggregateDone) {
                     if (isJobActive) {
                         HttpResponse httpResponse = this.getSubmission().getBatchSparkJobStatus(
                                 this.getConnectUri().toString(), batchId);
@@ -533,9 +533,7 @@ public Observable<SimpleImmutableEntry<SparkBatchJobState, String>> getJobDoneOb
                             isJobActive = !state.isJobDone();
                             applicationId = jobResp.getAppId();
                         }
-                    }
-
-                    if (!isLogAggregateDone && applicationId != null) {
+                    } else {
                         App yarnApp = this.getSparkJobYarnApplication(this.getConnectUri(), applicationId);
                         diagnostics = yarnApp.getDiagnostics();
 
@@ -555,13 +553,10 @@ public Observable<SimpleImmutableEntry<SparkBatchJobState, String>> getJobDoneOb
                     }
 
                     // Retry interval
-                    if (!isJobActive && isLogAggregateDone) {
-                        ob.onNext(new SimpleImmutableEntry<>(state, diagnostics));
-                        break;
-                    } else {
-                        sleep(1000);
-                    }
+                    sleep(1000);
                 }
+
+                ob.onNext(new SimpleImmutableEntry<>(state, diagnostics));
             } catch (IOException ex) {
                 ob.onError(ex);
             } catch (InterruptedException ignored) {