Fix job failure when using default blob storage account's credential for job upload storage

Author: t-rufang

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

@@ -93,6 +93,8 @@ protected SparkSubmissionParameter updateStorageConfigForSubmissionParameter(Spa
                                                       ? new SparkConfigures(existingConfigEntry)
                                                       : new SparkConfigures();
                 wrappedConfig.put("spark.hadoop." + fsRoot.getHadoopBlobFsPropertyKey(), storageKey);
+                // We need the following config to fix issue https://github.com/microsoft/azure-tools-for-java/issues/5002
+                wrappedConfig.put("spark.hadoop." + fsRoot.getKeyProviderPropertyKey(), fsRoot.getDefaultKeyProviderPropertyValue());
                 submissionParameter.getJobConfig().put(SparkSubmissionParameter.Conf, wrappedConfig);
             } catch (final UnknownFormatConversionException error) {
                 final String errorHint = "Azure blob storage uploading path is not in correct format";

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

@@ -25,6 +25,7 @@ public final class WasbUri extends AzureStorageUri {
             "^http[s]?://(?<storageAccount>[^/.]+)\\.blob\\.(?<endpointSuffix>[^/]+)/(?<container>[^/.]+)"
                     + "(:(?<port>[0-9]+))?(/(?<path>.*))?$",
             Pattern.CASE_INSENSITIVE);
+    public static final String SIMPLE_KEY_PROVIDER = "org.apache.hadoop.fs.azure.SimpleKeyProvider";
 
     private final LaterInit<String> container = new LaterInit<>();
     private final LaterInit<String> storageAccount = new LaterInit<>();
@@ -76,6 +77,14 @@ public String getHadoopBlobFsPropertyKey() {
         return String.format("fs.azure.account.key.%s.blob.%s", getStorageAccount(), getEndpointSuffix());
     }
 
+    public String getKeyProviderPropertyKey() {
+        return String.format("fs.azure.account.keyprovider.%s.blob.%s", getStorageAccount(), getEndpointSuffix());
+    }
+
+    public String getDefaultKeyProviderPropertyValue() {
+        return SIMPLE_KEY_PROVIDER;
+    }
+
     public static WasbUri parse(final String blobUri) {
         Matcher matcher;
         if (StringUtils.startsWithIgnoreCase(blobUri, "wasb")) {