Fixed a bug that caused failures when deleting empty directories (Azure#26674)

* Fixed a bug that caused failures when deleting empty directories

* Updated recordings

* Updated more recordings

Author: rickle-msft

sdk/storage/azure-storage-blob-nio/CHANGELOG.md

@@ -7,6 +7,7 @@
 ### Breaking Changes
 
 ### Bugs Fixed
+- Fixed a bug that would prevent deleting an empty directory in the case where one directory name was a prefix of the other.
 
 ### Other Changes
 

sdk/storage/azure-storage-blob-nio/src/main/java/com/azure/storage/blob/nio/AzureResource.java

@@ -10,6 +10,7 @@
 import com.azure.storage.blob.models.BlobHttpHeaders;
 import com.azure.storage.blob.models.BlobItem;
 import com.azure.storage.blob.models.BlobListDetails;
+import com.azure.storage.blob.models.BlobProperties;
 import com.azure.storage.blob.models.BlobRequestConditions;
 import com.azure.storage.blob.models.BlobStorageException;
 import com.azure.storage.blob.models.ListBlobsOptions;
@@ -93,43 +94,44 @@ DirectoryStatus checkDirStatus() throws IOException {
         }
         BlobContainerClient containerClient = this.getContainerClient();
 
-        // Two blobs will give us all the info we need (see below).
+        /*
+         * Do a get properties first on the directory name. This will determine if it is concrete&&exists or is either
+         * virtual or doesn't exist.
+         */
+        BlobProperties props = null;
+        boolean exists = false;
+        try {
+            props = this.getBlobClient().getProperties();
+            exists = true;
+        } catch (BlobStorageException e) {
+            if (e.getStatusCode() != 404) {
+                throw LoggingUtility.logError(logger, new IOException(e));
+            }
+        }
+
+        // Check if the resource is a file or directory before listing
+        if (exists && !props.getMetadata().containsKey(AzureResource.DIR_METADATA_MARKER)) {
+            return DirectoryStatus.NOT_A_DIRECTORY;
+        }
+
+        // List on the directory name + '/' so that we only get things under the directory if any
         ListBlobsOptions listOptions = new ListBlobsOptions().setMaxResultsPerPage(2)
-            .setPrefix(this.blobClient.getBlobName())
+            .setPrefix(this.blobClient.getBlobName() + AzureFileSystem.PATH_SEPARATOR)
             .setDetails(new BlobListDetails().setRetrieveMetadata(true));
 
         /*
-        Do a list on prefix.
-        Zero elements means no virtual dir. Does not exist.
-        One element that matches this dir means empty.
-        One element that doesn't match this dir or more than one element. Not empty.
-        One element that matches the name but does not have a directory marker means the resource is not a directory.
-
-        Note that blob names that match the prefix exactly are returned in listing operations.
+         * If listing returns anything, then it is not empty. If listing returns nothing and exists() was true, then it's
+         * empty Else it does not exist
          */
         try {
             Iterator<BlobItem> blobIterator = containerClient.listBlobsByHierarchy(AzureFileSystem.PATH_SEPARATOR,
                 listOptions, null).iterator();
-            if (!blobIterator.hasNext()) { // Nothing there
-                return DirectoryStatus.DOES_NOT_EXIST;
+            if (blobIterator.hasNext()) {
+                return DirectoryStatus.NOT_EMPTY;
+            } else if (exists) {
+                return DirectoryStatus.EMPTY;
             } else {
-                BlobItem item = blobIterator.next();
-                if (!item.getName().equals(this.blobClient.getBlobName())) {
-                    /*
-                    Names do not match. Must be a virtual dir with one item. e.g. blob with name "foo/bar" means dir
-                    "foo" exists.
-                     */
-                    return DirectoryStatus.NOT_EMPTY;
-                }
-                // Metadata marker
-                if (item.getMetadata() != null && item.getMetadata().containsKey(DIR_METADATA_MARKER)) {
-                    if (blobIterator.hasNext()) { // More than one item with dir path as prefix. Must be a dir.
-                        return DirectoryStatus.NOT_EMPTY;
-                    } else {
-                        return DirectoryStatus.EMPTY;
-                    }
-                }
-                return DirectoryStatus.NOT_A_DIRECTORY; // There is a file (not a directory) at this location.
+                return DirectoryStatus.DOES_NOT_EXIST;
             }
         } catch (BlobStorageException e) {
             throw LoggingUtility.logError(logger, new IOException(e));

sdk/storage/azure-storage-blob-nio/src/test/java/com/azure/storage/blob/nio/AzureFileSystemProviderTest.groovy

@@ -1481,11 +1481,15 @@ class AzureFileSystemProviderTest extends APISpec {
 
     def "CheckAccess IOException"() {
         setup:
-        config = initializeConfigMap(new CheckAccessIoExceptionPolicy())
         def fs = createFS(config)
         def path = fs.getPath(generateBlobName())
         def os = fs.provider().newOutputStream(path)
         os.close()
+        fs.close()
+
+        config = initializeConfigMap(new CheckAccessIoExceptionPolicy())
+        fs = createFS(config)
+        path = fs.getPath(path.toString())
 
         when:
         fs.provider().checkAccess(path)

sdk/storage/azure-storage-blob-nio/src/test/java/com/azure/storage/blob/nio/AzureResourceTest.groovy

@@ -105,7 +105,6 @@ class AzureResourceTest extends APISpec {
         DirectoryStatus.NOT_EMPTY       | false
     }
 
-    @Unroll
     def "Directory status files with same prefix"() {
         setup:
         def fs = createFS(config)
@@ -121,6 +120,40 @@ class AzureResourceTest extends APISpec {
         new AzureResource(path2).checkDirStatus() == DirectoryStatus.NOT_A_DIRECTORY
     }
 
+    def "Directory status directories with same prefix"() {
+        setup: "Create two folders where one is a prefix of the others"
+        def fs = createFS(config)
+        def pathName = generateBlobName()
+        def pathName2 = pathName + '2'
+        Files.createDirectory(fs.getPath(pathName))
+        Files.createDirectory(fs.getPath(pathName2))
+
+        expect: "Both should be empty"
+        new AzureResource(fs.getPath(pathName)).checkDirStatus() == DirectoryStatus.EMPTY
+        new AzureResource(fs.getPath(pathName2)).checkDirStatus() == DirectoryStatus.EMPTY
+    }
+
+    def "Directory status files between prefix and child"() {
+        setup:
+        def fs = createFS(config)
+        def dirPath = fs.getPath(generateBlobName())
+        def childPath = fs.getPath(dirPath.toString(), generateBlobName())
+        /*
+         Under an old listing scheme, it was possible for a file with the same name as a directory but with a trailing
+         '+' to cut in between the parent and child in the listing as we did it and the listing may not register the
+         child and erroneously return that the directory is empty. This ensures that listing is done in such a way as to
+         account for this and return correctly that the directory is not empty.
+         */
+        def middlePath = fs.getPath(dirPath.toString() + "+")
+
+        Files.createDirectory(dirPath)
+        Files.createFile(childPath)
+        Files.createFile(middlePath)
+
+        expect:
+        new AzureResource(dirPath).checkDirStatus() == DirectoryStatus.NOT_EMPTY
+    }
+
     def "Parent dir exists false"() {
         setup:
         def fs = createFS(config)

sdk/storage/azure-storage-blob-nio/src/test/java/com/azure/storage/blob/nio/CompositeTest.groovy

@@ -93,4 +93,17 @@ class CompositeTest extends APISpec {
         then:
         notThrown(IOException)
     }
+
+    def "Files delete empty directory"() {
+        setup: "Create two folders where one is a prefix of the others"
+        def fs = createFS(config)
+        def pathName = generateBlobName()
+        def pathName2 = pathName + '2'
+        Files.createDirectory(fs.getPath(pathName))
+        Files.createDirectory(fs.getPath(pathName2))
+
+        expect:
+        // Delete the one that is a prefix to ensure the other one does not interfere
+        Files.delete(fs.getPath(pathName))
+    }
 }