Fix cluster not exist for spark on ADL regression bug (#2197)

Author: t-rufang

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/azure/hdinsight/spark/ui/ServerlessSparkSubmissionPanelConfigurable.kt

@@ -30,10 +30,30 @@ import com.microsoft.azure.hdinsight.sdk.common.azure.serverless.AzureSparkServe
 import com.microsoft.azure.hdinsight.sdk.common.azure.serverless.AzureSparkServerlessClusterManager
 import com.microsoft.azure.hdinsight.spark.common.ServerlessSparkSubmitModel
 import com.microsoft.azure.hdinsight.spark.common.SparkSubmitModel
+import org.apache.commons.lang3.StringUtils
 import rx.Observable
+import java.util.stream.Collectors
 
 class ServerlessSparkSubmissionPanelConfigurable(project: Project, submissionPanel: SparkSubmissionContentPanel)
     : SparkSubmissionContentPanelConfigurable(project, submissionPanel), ILogger {
+
+    init {
+        jobUploadStorageCtrl = object : SparkSubmissionJobUploadStorageCtrl(storageWithUploadPathPanel) {
+            override fun getClusterName(): String? = selectedClusterDetail?.name
+
+            override fun getClusterDetail(): IClusterDetail? {
+                if (StringUtils.isEmpty(getClusterName())) {
+                    return null
+                }
+                return AzureSparkServerlessClusterManager.getInstance().clusters.stream()
+                        .filter { clusterDetail -> clusterDetail.name == getClusterName() }
+                        .collect(Collectors.toList())
+                        .getOrNull(0)
+            }
+        }
+
+    }
+
     override fun getType(): String = "Azure Data Lake Spark Pool"
 
     override fun getClusterDetails(): ImmutableSortedSet<out IClusterDetail> {

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/azure/hdinsight/spark/ui/SparkSubmissionContentPanelConfigurable.java

@@ -66,8 +66,8 @@ public class SparkSubmissionContentPanelConfigurable implements SettableControl<
     private final Project myProject;
 
     private SparkSubmissionContentPanel submissionPanel;
-    private SparkSubmissionJobUploadStorageWithUploadPathPanel storageWithUploadPathPanel;
-    private SparkSubmissionJobUploadStorageCtrl jobUploadStorageCtrl;
+    protected SparkSubmissionJobUploadStorageWithUploadPathPanel storageWithUploadPathPanel;
+    protected SparkSubmissionJobUploadStorageCtrl jobUploadStorageCtrl;
     private JPanel myWholePanel;
 
     // Cluster refresh publish subject with preselected cluster name as event
@@ -79,6 +79,7 @@ public SparkSubmissionContentPanelConfigurable(@NotNull Project project, @NotNul
         this.storageWithUploadPathPanel = submissionPanel.storageWithUploadPathPanel;
         this.myProject = project;
 
+/*
         this.jobUploadStorageCtrl = new SparkSubmissionJobUploadStorageCtrl(this.storageWithUploadPathPanel) {
             @Nullable
             @Override
@@ -87,6 +88,7 @@ public String getClusterName() {
                 return clusterDetail == null ? null : clusterDetail.getName();
             }
         };
+*/
         this.clustersRefreshSub = BehaviorSubject.create();
     }
 

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/azure/hdinsight/spark/ui/SparkSubmissionDebuggablePanelConfigurable.kt

@@ -24,6 +24,7 @@ package com.microsoft.azure.hdinsight.spark.ui
 
 import com.intellij.execution.configurations.RuntimeConfigurationError
 import com.intellij.openapi.project.Project
+import com.microsoft.azure.hdinsight.common.ClusterManagerEx
 import com.microsoft.azure.hdinsight.sdk.cluster.IClusterDetail
 import com.microsoft.azure.hdinsight.spark.common.SparkBatchRemoteDebugJobSshAuth.SSHAuthType.UsePassword
 import com.microsoft.azure.hdinsight.spark.common.SparkBatchRemoteDebugJobSshAuth.SSHAuthType.UseKeyFile
@@ -43,6 +44,19 @@ class SparkSubmissionDebuggablePanelConfigurable(project: Project,
         override fun getClusterNameToCheck(): String? = selectedClusterDetail?.name
     }
 
+    init {
+        jobUploadStorageCtrl = object : SparkSubmissionJobUploadStorageCtrl(storageWithUploadPathPanel) {
+            override fun getClusterName(): String? = selectedClusterDetail?.name
+
+            override fun getClusterDetail(): IClusterDetail? {
+                if (StringUtils.isEmpty(getClusterName())) {
+                    return null
+                }
+                return ClusterManagerEx.getInstance().getClusterDetailByName(getClusterName()).orElse(null)
+            }
+        }
+    }
+
     override fun onClusterSelected(cluster: IClusterDetail) {
         super.onClusterSelected(cluster)
 

PluginsAndFeatures/azure-toolkit-for-intellij/src/com/microsoft/azure/hdinsight/spark/ui/SparkSubmissionJobUploadStorageCtrl.kt

@@ -30,6 +30,7 @@ package com.microsoft.azure.hdinsight.spark.ui
 import com.microsoft.azure.hdinsight.common.ClusterManagerEx
 import com.microsoft.azure.hdinsight.common.logger.ILogger
 import com.microsoft.azure.hdinsight.sdk.cluster.IClusterDetail
+import com.microsoft.azure.hdinsight.sdk.common.azure.serverless.AzureSparkServerlessCluster
 import com.microsoft.azure.hdinsight.spark.common.SparkSubmitJobUploadStorageModel
 import com.microsoft.azure.hdinsight.spark.common.SparkSubmitStorageType
 import com.microsoft.tooling.msservices.helpers.azure.sdk.StorageClientSDKManager
@@ -104,12 +105,7 @@ abstract class SparkSubmissionJobUploadStorageCtrl(val view: SparkSubmissionJobU
                     { },
                     { err -> log().warn(ExceptionUtils.getStackTrace(err)) })
 
-    private fun getClusterDetail(): IClusterDetail? {
-        if (StringUtils.isEmpty(getClusterName())) {
-            return null
-        }
-        return ClusterManagerEx.getInstance().getClusterDetailByName(getClusterName()).orElse(null)
-    }
+    abstract fun getClusterDetail(): IClusterDetail?
 
     private fun validateStorageInfo(selectedItem: String): Observable<SparkSubmitJobUploadStorageModel> {
         return Observable.just(SparkSubmitJobUploadStorageModel())
@@ -136,10 +132,10 @@ abstract class SparkSubmissionJobUploadStorageCtrl(val view: SparkSubmissionJobU
                                 try {
                                     clusterDetail.getConfigurationInfo()
                                     val defaultStorageAccount = clusterDetail.storageAccount
-                                    val defaultStorageContainer = defaultStorageAccount.defaultContainerOrRootPath
-                                    if (defaultStorageAccount != null && defaultStorageContainer != null) {
+                                    val defaultContainerOrRootPath = defaultStorageAccount.defaultContainerOrRootPath
+                                    if (defaultStorageAccount != null && defaultContainerOrRootPath != null) {
                                         errorMsg = null
-                                        uploadPath = getAzureBlobStoragePath(defaultStorageAccount.name, defaultStorageContainer)
+                                        uploadPath = if (clusterDetail is AzureSparkServerlessCluster) getAzureDataLakeStoragePath(defaultContainerOrRootPath) else getAzureBlobStoragePath(defaultStorageAccount.name, defaultContainerOrRootPath)
                                         storageAccountType = SparkSubmitStorageType.DEFAULT_STORAGE_ACCOUNT
                                     } else {
                                         errorMsg = "Cluster have no storage account or storage container"
@@ -241,4 +237,13 @@ abstract class SparkSubmissionJobUploadStorageCtrl(val view: SparkSubmissionJobU
         return if (StringUtils.isEmpty(storageAccountName) || StringUtils.isEmpty(container)) null else
             "wasbs://$container@${ClusterManagerEx.getInstance().getBlobFullName(storageAccountName)}/SparkSubmission/"
     }
+
+    private fun getRootPathEndsWithSlash(rootPath: String): String {
+        return if (rootPath.endsWith("/")) rootPath else "$rootPath/"
+    }
+
+    private fun getAzureDataLakeStoragePath(rootPath: String?): String? {
+        return if (StringUtils.isEmpty(rootPath)) null else
+            "${getRootPathEndsWithSlash(rootPath.orEmpty())}SparkSubmission/"
+    }
 }