new test addition

Author: anmolanmol1234

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

@@ -21,9 +21,12 @@
 import java.io.IOException;
 import java.net.URI;
 import java.util.Hashtable;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 import org.assertj.core.api.Assertions;
 import org.junit.After;
@@ -700,4 +703,34 @@ protected void assertPathDns(Path path) {
         .describedAs("Path does not contain expected DNS")
         .contains(expectedDns);
   }
+
+  /**
+   * Checks a list of futures for exceptions.
+   *
+   * This method iterates over a list of futures, waits for each task to complete,
+   * and handles any exceptions thrown by the lambda expressions. If a
+   * RuntimeException is caught, it increments the exceptionCaught counter.
+   * If an unexpected exception is caught, it prints the exception to the standard error.
+   * Finally, it asserts that no RuntimeExceptions were caught.
+   *
+   * @param futures The list of futures to check for exceptions.
+   */
+  public void checkFuturesForExceptions(List<Future<?>> futures, int exceptionVal) {
+    int exceptionCaught = 0;
+    for (Future<?> future : futures) {
+      try {
+        future.get(); // wait for the task to complete and handle any exceptions thrown by the lambda expression
+      } catch (ExecutionException e) {
+        Throwable cause = e.getCause();
+        if (cause instanceof RuntimeException) {
+          exceptionCaught++;
+        } else {
+          System.err.println("Unexpected exception caught: " + cause);
+        }
+      } catch (InterruptedException e) {
+        // handle interruption
+      }
+    }
+    assertEquals(exceptionCaught, exceptionVal);
+  }
 }

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

@@ -29,7 +29,6 @@
 import java.util.List;
 import java.util.Random;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -510,36 +509,6 @@ public void testRecreateDirectoryAppendAndFlush() throws IOException {
     }
   }
 
-  /**
-   * Checks a list of futures for exceptions.
-   *
-   * This method iterates over a list of futures, waits for each task to complete,
-   * and handles any exceptions thrown by the lambda expressions. If a
-   * RuntimeException is caught, it increments the exceptionCaught counter.
-   * If an unexpected exception is caught, it prints the exception to the standard error.
-   * Finally, it asserts that no RuntimeExceptions were caught.
-   *
-   * @param futures The list of futures to check for exceptions.
-   */
-  private void checkFuturesForExceptions(List<Future<?>> futures, int exceptionVal) {
-    int exceptionCaught = 0;
-    for (Future<?> future : futures) {
-      try {
-        future.get(); // wait for the task to complete and handle any exceptions thrown by the lambda expression
-      } catch (ExecutionException e) {
-        Throwable cause = e.getCause();
-        if (cause instanceof RuntimeException) {
-          exceptionCaught++;
-        } else {
-          System.err.println("Unexpected exception caught: " + cause);
-        }
-      } catch (InterruptedException e) {
-        // handle interruption
-      }
-    }
-    assertEquals(exceptionCaught, exceptionVal);
-  }
-
   /**
    * Verify that parallel write with same offset from different output streams will not throw exception.
    **/

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

@@ -25,11 +25,13 @@
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 
 import org.assertj.core.api.Assertions;
 import org.junit.Assume;
@@ -72,6 +74,7 @@
 import static java.net.HttpURLConnection.HTTP_PRECON_FAILED;
 import static org.apache.hadoop.fs.azurebfs.AbfsStatistic.CONNECTIONS_MADE;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.ROOT_PATH;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_CONDITIONAL_CREATE_OVERWRITE;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_MKDIR_OVERWRITE;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_MB;
 import static org.apache.hadoop.fs.azurebfs.services.RenameAtomicity.SUFFIX;
@@ -925,6 +928,96 @@ public void testCreateSubPath() throws Exception {
     intercept(IOException.class, () -> fs.create(new Path("a/b")));
   }
 
+  /**
+   * Test create path in parallel with overwrite false.
+   **/
+  @Test
+  public void testParallelCreateOverwriteFalse()
+      throws Exception {
+    Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_ENABLE_CONDITIONAL_CREATE_OVERWRITE, "false");
+    try (AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(
+        configuration)) {
+      ExecutorService executorService = Executors.newFixedThreadPool(5);
+      List<Future<?>> futures = new ArrayList<>();
+
+      final byte[] b = new byte[8 * ONE_MB];
+      new Random().nextBytes(b);
+      final Path filePath = path("/testPath");
+
+      futures.add(executorService.submit(() -> {
+        try {
+          fs.create(filePath, false);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }));
+
+      futures.add(executorService.submit(() -> {
+        try {
+          fs.create(filePath, false);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }));
+
+      futures.add(executorService.submit(() -> {
+        try {
+          fs.create(filePath, false);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }));
+
+      checkFuturesForExceptions(futures, 2);
+    }
+  }
+
+  /**
+   * Test create path in parallel with overwrite true.
+   **/
+  @Test
+  public void testParallelCreateOverwriteTrue()
+      throws Exception {
+    Configuration configuration = getRawConfiguration();
+    configuration.set(FS_AZURE_ENABLE_CONDITIONAL_CREATE_OVERWRITE, "false");
+    try (AzureBlobFileSystem fs = (AzureBlobFileSystem) FileSystem.newInstance(
+        configuration)) {
+      ExecutorService executorService = Executors.newFixedThreadPool(5);
+      List<Future<?>> futures = new ArrayList<>();
+
+      final byte[] b = new byte[8 * ONE_MB];
+      new Random().nextBytes(b);
+      final Path filePath = path("/testPath");
+
+      futures.add(executorService.submit(() -> {
+        try {
+          fs.create(filePath);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }));
+
+      futures.add(executorService.submit(() -> {
+        try {
+          fs.create(filePath);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }));
+
+      futures.add(executorService.submit(() -> {
+        try {
+          fs.create(filePath);
+        } catch (IOException e) {
+          throw new RuntimeException(e);
+        }
+      }));
+
+      checkFuturesForExceptions(futures, 0);
+    }
+  }
+
   /**
    * Creating path with parent explicit.
    */