Fix the Spark 2.1 cluster timing issue for remote debugging.

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

Author: wezhang

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

@@ -62,6 +62,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.net.UnknownServiceException;
+import java.util.AbstractMap;
 import java.util.AbstractMap.SimpleEntry;
 import java.util.HashMap;
 import java.util.Optional;
@@ -138,7 +139,7 @@ protected void execute(@NotNull ExecutionEnvironment environment, @Nullable Call
         debugProcessPhaser = new Phaser(1);
 
         Observable.create((Observable.OnSubscribe<String>) ob ->
-                createDebugJobSession(submitModel).subscribe(debugJobClusterPair -> {
+                createDebugJobSession(submitModel, ob).subscribe(debugJobClusterPair -> {
                     final SparkBatchRemoteDebugJob remoteDebugJob = debugJobClusterPair.getKey();
                     final IClusterDetail clusterDetail = debugJobClusterPair.getValue();
                     final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
@@ -176,16 +177,9 @@ protected void execute(@NotNull ExecutionEnvironment environment, @Nullable Call
 
                     ob.onNext("Info: Spark Job Driver debugging started.");
 
-                    Subscription livyLogSubscription = submitModel
-                            .jobLogObservable(remoteDebugJob.getBatchId(), clusterDetail)
-                            .subscribeOn(Schedulers.io())
-                            .subscribe();
-
                     // Await for all debug processes finish
                     debugProcessPhaser.arriveAndAwaitAdvance();
                     ob.onCompleted();
-
-                    livyLogSubscription.unsubscribe();
                 }, ob::onError))
                 .subscribe(
                         info -> HDInsightUtil.showInfoOnSubmissionMessageWindow(project, info),
@@ -252,7 +246,7 @@ private ExecutionEnvironment buildChildEnvironment(@NotNull ExecutionEnvironment
      * Create a Debug Spark Job session with building, deploying and submitting
      */
     private Single<SimpleEntry<SparkBatchRemoteDebugJob, IClusterDetail>> createDebugJobSession(
-                                                                            @NotNull SparkSubmitModel submitModel) {
+            @NotNull SparkSubmitModel submitModel, Subscriber<? super String> debugSessionSub) {
         SparkSubmissionParameter submissionParameter = submitModel.getSubmissionParameter();
         String selectedClusterName = submissionParameter.getClusterName();
 
@@ -285,11 +279,23 @@ private Single<SimpleEntry<SparkBatchRemoteDebugJob, IClusterDetail>> createDebu
                                 submitModel.tryToCreateBatchSparkDebugJob(selectedClusterDetail);
                         setDebugJob(remoteDebugJob);
 
+                        return new SimpleEntry<>(remoteDebugJob, selectedClusterDetail);
+                    } catch (Exception e) {
+                        HDInsightUtil.setJobRunningStatus(submitModel.getProject(), false);
+                        throw propagate(e);
+                    }})
+                .flatMap(jobClusterPair -> jobClusterPair.getKey().getSubmissionLog()
+                                                         .map(AbstractMap.SimpleImmutableEntry::getValue)
+                                                         .doOnNext(debugSessionSub::onNext)
+                                                         .doOnError(debugSessionSub::onError)
+                                                         .last()
+                                                         .toSingle()
+                                                         .map(message -> jobClusterPair))
+                .doOnSuccess(jobClusterPair -> {
+                    try {
                         SparkBatchDebugSession session = createSparkBatchDebugSession(
-                                selectedClusterDetail.getConnectionUrl(), submitModel.getAdvancedConfigModel()).open();
+                                jobClusterPair.getValue().getConnectionUrl(), submitModel.getAdvancedConfigModel()).open();
                         setDebugSession(session);
-
-                        return new SimpleEntry<>(remoteDebugJob, selectedClusterDetail);
                     } catch (Exception e) {
                         HDInsightUtil.setJobRunningStatus(submitModel.getProject(), false);
                         throw propagate(e);

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

@@ -638,6 +638,7 @@ public Observable<SimpleImmutableEntry<MessageInfoType, String>> getSubmissionLo
                 boolean isSubmitting = true;
 
                 while (isSubmitting) {
+                    Boolean isAppIdAllocated = !this.getSparkJobApplicationIdObservable().isEmpty().toBlocking().last();
                     String logUrl = String.format("%s/%d/log?from=%d&size=%d",
                             this.getConnectUri().toString(), batchId, start, maxLinesPerGet);
 
@@ -656,8 +657,9 @@ public Observable<SimpleImmutableEntry<MessageInfoType, String>> getSubmissionLo
 
                     // Retry interval
                     if (linesGot == 0) {
+                        isSubmitting = this.getState().equals("starting") && !isAppIdAllocated;
+
                         sleep(TimeUnit.SECONDS.toMillis(this.getDelaySeconds()));
-                        isSubmitting = this.getState().equals("starting");
                     }
                 }
             } catch (IOException ex) {