Fix S3 delete throwing DirectoryNotEmptyException due to eventual consistency (#6833)

Signed-off-by: Paolo Di Tommaso <paolo.ditommaso@gmail.com>

Author: pditommaso

plugins/nf-amazon/src/main/nextflow/cloud/aws/nio/S3FileSystemProvider.java

@@ -485,13 +485,11 @@ public void delete(Path path) throws IOException {
             throw new NoSuchFileException("the path: " + FilesEx.toUriString(s3Path) + " does not exist");
         }
 
-        if (Files.isDirectory(path)){
-            try (DirectoryStream<Path> stream = Files.newDirectoryStream(path)){
-                if (stream.iterator().hasNext()){
-                    throw new DirectoryNotEmptyException("the path: " + FilesEx.toUriString(s3Path) + " is a directory and is not empty");
-                }
-            }
-        }
+		// NOTE: S3 directories are virtual (marker objects or implied key prefixes),
+		// so we do not check for emptiness before deleting. Enforcing POSIX-like
+		// DirectoryNotEmptyException semantics on S3 is unreliable due to eventual
+		// consistency and unnecessary because deleting a directory marker does not
+		// affect its children.
 
 		// we delete the two objects (sometimes exists the key '/' and sometimes not)
 		s3Path.getFileSystem().getClient()

plugins/nf-amazon/src/test/nextflow/cloud/aws/nio/AwsS3NioTest.groovy

@@ -495,6 +495,24 @@ class AwsS3NioTest extends Specification implements AwsS3BaseSpec {
 
     }
 
+    def 'should delete a non-empty directory on S3 without throwing' () {
+        given:
+        def bucketName = createBucket()
+        and:
+        createObject("$bucketName/dir1/file1.txt", 'HELLO')
+        createObject("$bucketName/dir1/file2.txt", 'WORLD')
+
+        when:
+        def path = s3path("s3://$bucketName/dir1")
+        Files.delete(path)
+
+        then:
+        noExceptionThrown()
+
+        cleanup:
+        deleteBucket(bucketName)
+    }
+
     @Ignore //FIXME
     def 'should validate exists method' () {
         given: