HADOOP-14461 Azure: handle failure gracefully in case of missing account access key.

Contributed by Mingliang Liu.

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureNativeFileSystemStore.java

@@ -303,6 +303,14 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
   private boolean useLocalSasKeyMode = false;
 
   private String delegationToken;
+
+  /** The error message template when container is not accessible. */
+  static final String NO_ACCESS_TO_CONTAINER_MSG = "No credentials found for "
+      + "account %s in the configuration, and its container %s is not "
+      + "accessible using anonymous credentials. Please check if the container "
+      + "exists first. If it is not publicly available, you have to provide "
+      + "account credentials.";
+
   /**
    * A test hook interface that can modify the operation context we use for
    * Azure Storage operations, e.g. to inject errors.
@@ -778,18 +786,17 @@ private void connectUsingAnonymousCredentials(final URI uri)
     rootDirectory = container.getDirectoryReference("");
 
     // Check for container existence, and our ability to access it.
+    boolean canAccess;
     try {
-      if (!container.exists(getInstrumentedContext())) {
-        throw new AzureException("Container " + containerName + " in account "
-            + accountName + " not found, and we can't create"
-            + " it using anoynomous credentials, and no credentials found for them"
-            + " in the configuration.");
-      }
+      canAccess = container.exists(getInstrumentedContext());
     } catch (StorageException ex) {
-      throw new AzureException("Unable to access container " + containerName
-          + " in account " + accountName
-          + " using anonymous credentials, and no credentials found for them "
-          + " in the configuration.", ex);
+      LOG.error("Service returned StorageException when checking existence "
+          + "of container {} in account {}", containerName, accountName, ex);
+      canAccess = false;
+    }
+    if (!canAccess) {
+      throw new AzureException(String.format(NO_ACCESS_TO_CONTAINER_MSG,
+          accountName, containerName));
     }
 
     // Accessing the storage server unauthenticated using
@@ -999,22 +1006,17 @@ private void createAzureStorageSession ()
       // Check whether the account is configured with an account key.
       propertyValue = getAccountKeyFromConfiguration(accountName,
           sessionConfiguration);
-      if (propertyValue != null) {
-
+      if (StringUtils.isNotEmpty(propertyValue)) {
         // Account key was found.
         // Create the Azure storage session using the account key and container.
         connectUsingConnectionStringCredentials(
             getAccountFromAuthority(sessionUri),
             getContainerFromAuthority(sessionUri), propertyValue);
-
-        // Return to caller
-        return;
+      } else {
+        LOG.debug("The account access key is not configured for {}. "
+            + "Now try anonymous access.", sessionUri);
+        connectUsingAnonymousCredentials(sessionUri);
       }
-
-      // The account access is not configured for this cluster. Try anonymous
-      // access.
-      connectUsingAnonymousCredentials(sessionUri);
-
     } catch (Exception e) {
       // Caught exception while attempting to initialize the Azure File
       // System store, re-throw the exception.

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/AzureBlobStorageTestAccount.java

@@ -22,6 +22,9 @@
 import com.microsoft.azure.storage.blob.*;
 import com.microsoft.azure.storage.core.Base64;
 import org.apache.commons.configuration2.SubsetConfiguration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.azure.metrics.AzureFileSystemInstrumentation;
@@ -46,6 +49,8 @@
  * for instructions on how to connect to a real Azure Storage account.
  */
 public final class AzureBlobStorageTestAccount {
+  private static final Logger LOG = LoggerFactory.getLogger(
+      AzureBlobStorageTestAccount.class);
 
   private static final String ACCOUNT_KEY_PROPERTY_NAME = "fs.azure.account.key.";
   private static final String SAS_PROPERTY_NAME = "fs.azure.sas.";
@@ -299,10 +304,9 @@ public static AzureBlobStorageTestAccount createForEmulator()
     Configuration conf = createTestConfiguration();
     if (!conf.getBoolean(USE_EMULATOR_PROPERTY_NAME, false)) {
       // Not configured to test against the storage emulator.
-      System.out
-        .println("Skipping emulator Azure test because configuration " +
-            "doesn't indicate that it's running." +
-            " Please see RunningLiveWasbTests.txt for guidance.");
+      LOG.warn("Skipping emulator Azure test because configuration doesn't "
+          + "indicate that it's running. Please see RunningLiveWasbTests.txt "
+          + "for guidance.");
       return null;
     }
     CloudStorageAccount account =
@@ -456,18 +460,22 @@ static CloudStorageAccount createStorageAccount(String accountName,
       KeyProviderException {
     String accountKey = AzureNativeFileSystemStore
         .getAccountKeyFromConfiguration(accountName, conf);
-    StorageCredentials credentials;
-    if (accountKey == null && allowAnonymous) {
-      credentials = StorageCredentialsAnonymous.ANONYMOUS;
+    final StorageCredentials credentials;
+    if (accountKey == null) {
+      if (allowAnonymous) {
+        credentials = StorageCredentialsAnonymous.ANONYMOUS;
+      } else {
+        LOG.warn("Skipping live Azure test because of missing key for"
+            + " account '" + accountName + "'. "
+            + "Please see RunningLiveWasbTests.txt for guidance.");
+        return null;
+      }
     } else {
       credentials = new StorageCredentialsAccountAndKey(
           accountName.split("\\.")[0], accountKey);
     }
-    if (credentials == null) {
-      return null;
-    } else {
-      return new CloudStorageAccount(credentials);
-    }
+
+    return new CloudStorageAccount(credentials);
   }
 
   public static Configuration createTestConfiguration() {
@@ -493,9 +501,8 @@ static CloudStorageAccount createTestAccount(Configuration conf)
       throws URISyntaxException, KeyProviderException {
     String testAccountName = conf.get(TEST_ACCOUNT_NAME_PROPERTY_NAME);
     if (testAccountName == null) {
-      System.out
-        .println("Skipping live Azure test because of missing test account." +
-                 " Please see RunningLiveWasbTests.txt for guidance.");
+      LOG.warn("Skipping live Azure test because of missing test account. "
+          + "Please see RunningLiveWasbTests.txt for guidance.");
       return null;
     }
     return createStorageAccount(testAccountName, conf, false);

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestAzureFileSystemErrorConditions.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.fs.azure;
 
+import static org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.NO_ACCESS_TO_CONTAINER_MSG;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
@@ -35,6 +36,8 @@
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.TestHookOperationContext;
+import org.apache.hadoop.test.GenericTestUtils;
+
 import org.junit.Test;
 
 import com.microsoft.azure.storage.OperationContext;
@@ -64,18 +67,18 @@ public void testNoInitialize() throws Exception {
    */
   @Test
   public void testAccessUnauthorizedPublicContainer() throws Exception {
+    final String container = "nonExistentContainer";
+    final String account = "hopefullyNonExistentAccount";
     Path noAccessPath = new Path(
-        "wasb://nonExistentContainer@hopefullyNonExistentAccount/someFile");
+        "wasb://" + container + "@" + account + "/someFile");
     NativeAzureFileSystem.suppressRetryPolicy();
     try {
       FileSystem.get(noAccessPath.toUri(), new Configuration())
         .open(noAccessPath);
       assertTrue("Should've thrown.", false);
     } catch (AzureException ex) {
-      assertTrue("Unexpected message in exception " + ex,
-          ex.getMessage().contains(
-          "Unable to access container nonExistentContainer in account" +
-          " hopefullyNonExistentAccount"));
+      GenericTestUtils.assertExceptionContains(
+          String.format(NO_ACCESS_TO_CONTAINER_MSG, account, container), ex);
     } finally {
       NativeAzureFileSystem.resumeRetryPolicy();
     }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestFileSystemOperationExceptionMessage.java

@@ -21,9 +21,13 @@
 import java.util.UUID;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.test.GenericTestUtils;
+
 import org.junit.Assert;
 import org.junit.Test;
 
+import static org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.NO_ACCESS_TO_CONTAINER_MSG;
+
 
 public class TestFileSystemOperationExceptionMessage extends
   NativeAzureFileSystemBaseTest {
@@ -41,11 +45,6 @@ public void testAnonymouseCredentialExceptionMessage() throws Throwable{
     String wasbUri = String.format("wasb://%s@%s",
         testContainer, testStorageAccount);
 
-    String expectedErrorMessage =
-        String.format("Container %s in account %s not found, and we can't create it "
-            + "using anoynomous credentials, and no credentials found for "
-            + "them in the configuration.", testContainer, testStorageAccount);
-
     fs = new NativeAzureFileSystem();
     try {
       fs.initialize(new URI(wasbUri), conf);
@@ -63,7 +62,9 @@ public void testAnonymouseCredentialExceptionMessage() throws Throwable{
             || exceptionMessage.length() == 0) {
           Assert.fail();}
         else {
-          Assert.assertTrue(exceptionMessage.equals(expectedErrorMessage));
+          GenericTestUtils.assertExceptionContains(String.format(
+              NO_ACCESS_TO_CONTAINER_MSG, testStorageAccount, testContainer),
+              ex);
         }
       } else {
         Assert.fail();