HADOOP-14583. wasb throws an exception if you try to create a file and there's no parent directory.

Contributed by Esfandiar Manii.

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

@@ -2077,6 +2077,7 @@ public FileMetadata retrieveMetadata(String key) throws IOException {
 
         LOG.debug("Found {} as an explicit blob. Checking if it's a file or folder.", key);
 
+        try {
           // The blob exists, so capture the metadata from the blob
           // properties.
           blob.downloadAttributes(getInstrumentedContext());
@@ -2096,6 +2097,11 @@ public FileMetadata retrieveMetadata(String key) throws IOException {
                 properties.getLastModified().getTime(),
                 getPermissionStatus(blob));
           }
+        } catch(StorageException e){
+          if (!NativeAzureFileSystemHelper.isFileNotFoundException(e)) {
+            throw e;
+          }
+        }
       }
 
       // There is no file with that key name, but maybe it is a folder.

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

@@ -19,101 +19,169 @@
 package org.apache.hadoop.fs.azure;
 
 
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
 /***
  * Test class to hold all Live Azure storage concurrency tests.
  */
 public class TestNativeAzureFileSystemConcurrencyLive
     extends AbstractWasbTestBase {
 
-  private static final int TEST_COUNT = 102;
+  private static final int THREAD_COUNT = 102;
+  private static final int TEST_EXECUTION_TIMEOUT = 5000;
+
   @Override
   protected AzureBlobStorageTestAccount createTestAccount() throws Exception {
     return AzureBlobStorageTestAccount.create();
   }
 
   /**
-   * Test multi-threaded deletes in WASB. Expected behavior is one of the thread
+   * Validate contract for FileSystem.create when overwrite is true and there
-   * should be to successfully delete the file and return true and all other
+   * are concurrent callers of FileSystem.delete.  An existing file should be
-   * threads need to return false.
+   * overwritten, even if the original destination exists but is deleted by an
+   * external agent during the create operation.
    */
-  @Test
+  @Test(timeout = TEST_EXECUTION_TIMEOUT)
-  public void testMultiThreadedDeletes() throws Exception {
+  public void testConcurrentCreateDeleteFile() throws Exception {
     Path testFile = new Path("test.dat");
     fs.create(testFile).close();
 
-    int threadCount = TEST_COUNT;
+    List<CreateFileTask> tasks = new ArrayList<>(THREAD_COUNT);
-    DeleteHelperThread[] helperThreads = new DeleteHelperThread[threadCount];
 
-    for (int i = 0; i < threadCount; i++) {
+    for (int i = 0; i < THREAD_COUNT; i++) {
-      helperThreads[i] = new DeleteHelperThread(fs, testFile);
+      tasks.add(new CreateFileTask(fs, testFile));
     }
 
-    Thread[] threads = new Thread[threadCount];
+    ExecutorService es = null;
 
-    for (int i = 0; i < threadCount; i++) {
-      threads[i] = new Thread(helperThreads[i]);
-      threads[i].start();
-    }
-
-    for (int i = 0; i < threadCount; i++) {
-      threads[i].join();
-    }
-
-    boolean deleteSuccess = false;
-
-    for (int i = 0; i < threadCount; i++) {
-
-      Assert.assertFalse("child thread has exception : " + helperThreads[i].getException(),
-          helperThreads[i].getExceptionEncounteredFlag());
-
-      if (deleteSuccess) {
-        Assert.assertFalse("More than one thread delete() retuhelperThreads[i].getDeleteSuccess()",
-            helperThreads[i].getExceptionEncounteredFlag());
-      } else {
-        deleteSuccess = helperThreads[i].getDeleteSuccess();
-      }
-    }
-
-    Assert.assertTrue("No successfull delete found", deleteSuccess);
-  }
-}
-
-class DeleteHelperThread implements Runnable {
-
-  private FileSystem fs;
-  private Path p;
-  private boolean deleteSuccess;
-  private boolean exceptionEncountered;
-  private Exception ex;
-
-  public DeleteHelperThread(FileSystem fs, Path p) {
-    this.fs = fs;
-    this.p = p;
-  }
-
-  public void run() {
     try {
-      deleteSuccess = fs.delete(p, false);
+      es = Executors.newFixedThreadPool(THREAD_COUNT);
-    } catch (Exception ioEx) {
+
-      exceptionEncountered = true;
+      List<Future<Void>> futures = es.invokeAll(tasks);
-      this.ex = ioEx;
+
+      for (Future<Void> future : futures) {
+        Assert.assertTrue(future.isDone());
+
+        // we are using Callable<V>, so if an exception
+        // occurred during the operation, it will be thrown
+        // when we call get
+        Assert.assertEquals(null, future.get());
+      }
+    } finally {
+      if (es != null) {
+        es.shutdownNow();
+      }
     }
   }
 
-  public boolean getDeleteSuccess() {
+  /**
-    return deleteSuccess;
+   * Validate contract for FileSystem.delete when invoked concurrently.
+   * One of the threads should successfully delete the file and return true;
+   * all other threads should return false.
+   */
+  @Test(timeout = TEST_EXECUTION_TIMEOUT)
+  public void testConcurrentDeleteFile() throws Exception {
+    Path testFile = new Path("test.dat");
+    fs.create(testFile).close();
+
+    List<DeleteFileTask> tasks = new ArrayList<>(THREAD_COUNT);
+
+    for (int i = 0; i < THREAD_COUNT; i++) {
+      tasks.add(new DeleteFileTask(fs, testFile));
     }
 
-  public boolean getExceptionEncounteredFlag() {
+    ExecutorService es = null;
-    return exceptionEncountered;
+    try {
+      es = Executors.newFixedThreadPool(THREAD_COUNT);
+
+      List<Future<Boolean>> futures = es.invokeAll(tasks);
+
+      int successCount = 0;
+      for (Future<Boolean> future : futures) {
+        Assert.assertTrue(future.isDone());
+
+        // we are using Callable<V>, so if an exception
+        // occurred during the operation, it will be thrown
+        // when we call get
+        Boolean success = future.get();
+        if (success) {
+          successCount++;
+        }
       }
 
-  public Exception getException() {
+      Assert.assertEquals(
-    return ex;
+          "Exactly one delete operation should return true.",
+          1,
+          successCount);
+    } finally {
+      if (es != null) {
+        es.shutdownNow();
+      }
+    }
+  }
+
+
+  abstract class FileSystemTask<V> implements Callable<V> {
+    private final FileSystem fileSystem;
+    private final Path path;
+
+    protected FileSystem getFileSystem() {
+      return this.fileSystem;
+    }
+
+    protected Path getFilePath() {
+      return this.path;
+    }
+
+    FileSystemTask(FileSystem fs, Path p) {
+      this.fileSystem = fs;
+      this.path = p;
+    }
+
+    public abstract V call() throws Exception;
+  }
+
+  class DeleteFileTask extends FileSystemTask<Boolean> {
+
+    DeleteFileTask(FileSystem fs, Path p) {
+      super(fs, p);
+    }
+
+    @Override
+    public Boolean call() throws Exception {
+      return this.getFileSystem().delete(this.getFilePath(), false);
+    }
+  }
+
+  class CreateFileTask extends FileSystemTask<Void> {
+    CreateFileTask(FileSystem fs, Path p) {
+      super(fs, p);
+    }
+
+    public Void call() throws Exception {
+      FileSystem fs = getFileSystem();
+      Path p = getFilePath();
+
+      // Create an empty file and close the stream.
+      FSDataOutputStream stream = fs.create(p, true);
+      stream.close();
+
+      // Delete the file.  We don't care if delete returns true or false.
+      // We just want to ensure the file does not exist.
+      this.getFileSystem().delete(this.getFilePath(), false);
+
+      return null;
+    }
   }
 }