HADOOP-19464. S3A: Restore Compatibility with EMRFS FileSystem

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Listing.java

@@ -50,6 +50,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.StringJoiner;
 
+import static org.apache.hadoop.fs.s3a.Constants.S3N_FOLDER_SUFFIX;
 import static org.apache.hadoop.fs.s3a.Invoker.onceInTheFuture;
 import static org.apache.hadoop.fs.s3a.S3AUtils.createFileStatus;
 import static org.apache.hadoop.fs.s3a.S3AUtils.maybeAddTrailingSlash;
@@ -74,6 +75,9 @@ public class Listing extends AbstractStoreOperation {
 
   private static final Logger LOG = S3AFileSystem.LOG;
 
+  static final FileStatusAcceptor ACCEPT_ALL_BUT_S3N =
+      new AcceptAllButS3nDirs();
+
   static final FileStatusAcceptor ACCEPT_ALL_OBJECTS =
       new AcceptAllObjects();
 
@@ -114,7 +118,7 @@ public static RemoteIterator<S3AFileStatus> toProvidedFileStatusIterator(
           S3AFileStatus[] fileStatuses) {
     return filteringRemoteIterator(
         remoteIteratorFromArray(fileStatuses),
-        Listing.ACCEPT_ALL_OBJECTS::accept);
+        Listing.ACCEPT_ALL_BUT_S3N::accept);
   }
 
   /**
@@ -233,7 +237,7 @@ public RemoteIterator<S3ALocatedFileStatus> getLocatedFileStatusIteratorForDir(
             listingOperationCallbacks
                 .createListObjectsRequest(key, "/", span),
             filter,
-            new AcceptAllButSelf(dir),
+            new AcceptAllButSelfAndS3nDirs(dir),
             span));
   }
 
@@ -262,7 +266,7 @@ public RemoteIterator<S3ALocatedFileStatus> getLocatedFileStatusIteratorForDir(
         path,
         request,
         S3AUtils.ACCEPT_ALL,
-        new AcceptAllButSelf(path),
+        new AcceptAllButSelfAndS3nDirs(path),
         span);
   }
 
@@ -460,7 +464,7 @@ private boolean buildNextStatusBatch(S3ListResult objects) throws IOException {
         if (LOG.isDebugEnabled()) {
           LOG.debug("{}: {}", keyPath, stringify(s3Object));
         }
-        // Skip over keys that are ourselves
+        // Skip over keys that are ourselves and old S3N _$folder$ files
         if (acceptor.accept(keyPath, s3Object) && filter.accept(keyPath)) {
           S3AFileStatus status = createFileStatus(keyPath, s3Object,
               blockSize, userName, s3Object.eTag(),
@@ -720,7 +724,8 @@ public void close() {
   }
 
   /**
-   * Accept all entries except the base path.
+   * Accept all entries except the base path and those which map to S3N
+   * pseudo directory markers.
    */
   static class AcceptFilesOnly implements FileStatusAcceptor {
     private final Path qualifiedPath;
@@ -740,6 +745,7 @@ static class AcceptFilesOnly implements FileStatusAcceptor {
     @Override
     public boolean accept(Path keyPath, S3Object s3Object) {
       return !keyPath.equals(qualifiedPath)
+          && !s3Object.key().endsWith(S3N_FOLDER_SUFFIX)
           && !objectRepresentsDirectory(s3Object.key());
     }
 
@@ -760,6 +766,25 @@ public boolean accept(FileStatus status) {
     }
   }
 
+  /**
+   * Accept all entries except those which map to S3N pseudo directory markers.
+   */
+  static class AcceptAllButS3nDirs implements FileStatusAcceptor {
+
+    public boolean accept(Path keyPath, S3Object s3Object) {
+      return !s3Object.key().endsWith(S3N_FOLDER_SUFFIX);
+    }
+
+    public boolean accept(Path keyPath, String prefix) {
+      return !keyPath.toString().endsWith(S3N_FOLDER_SUFFIX);
+    }
+
+    public boolean accept(FileStatus status) {
+      return !status.getPath().toString().endsWith(S3N_FOLDER_SUFFIX);
+    }
+
+  }
+
   /**
    * Accept all entries.
    */
@@ -779,6 +804,55 @@ public boolean accept(FileStatus status) {
 
   }
 
+  /**
+   * Accept all entries except the base path and those which map to S3N
+   * pseudo directory markers.
+   */
+  public static class AcceptAllButSelfAndS3nDirs implements FileStatusAcceptor {
+
+    /** Base path. */
+    private final Path qualifiedPath;
+
+    /**
+     * Constructor.
+     * @param qualifiedPath an already-qualified path.
+     */
+    public AcceptAllButSelfAndS3nDirs(Path qualifiedPath) {
+      this.qualifiedPath = qualifiedPath;
+    }
+
+    /**
+     * Reject a s3Object entry if the key path is the qualified Path, or
+     * it ends with {@code "_$folder$"}.
+     * @param keyPath key path of the entry
+     * @param s3Object s3Object entry
+     * @return true if the entry is accepted (i.e. that a status entry
+     * should be generated.)
+     */
+    @Override
+    public boolean accept(Path keyPath, S3Object s3Object) {
+      return !keyPath.equals(qualifiedPath) &&
+          !s3Object.key().endsWith(S3N_FOLDER_SUFFIX);
+    }
+
+    /**
+     * Accept all prefixes except the one for the base path, "self".
+     * @param keyPath qualified path to the entry
+     * @param prefix common prefix in listing.
+     * @return true if the entry is accepted (i.e. that a status entry
+     * should be generated.
+     */
+    @Override
+    public boolean accept(Path keyPath, String prefix) {
+      return !keyPath.equals(qualifiedPath);
+    }
+
+    @Override
+    public boolean accept(FileStatus status) {
+      return (status != null) && !status.getPath().equals(qualifiedPath);
+    }
+  }
+
   /**
    * Accept all entries except the base path.
    */

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -2651,7 +2651,7 @@ public RemoteIterator<S3AFileStatus> listObjects(
           listing.createFileStatusListingIterator(path,
               createListObjectsRequest(key, null),
               ACCEPT_ALL,
-              Listing.ACCEPT_ALL_OBJECTS,
+              Listing.ACCEPT_ALL_BUT_S3N,
               auditSpan));
     }
 
@@ -3679,7 +3679,7 @@ private RemoteIterator<S3AFileStatus> innerListStatus(Path f)
         return listing.createProvidedFileStatusIterator(
                 stats,
                 ACCEPT_ALL,
-                Listing.ACCEPT_ALL_OBJECTS);
+                Listing.ACCEPT_ALL_BUT_S3N);
       }
     }
     // Here we have a directory which may or may not be empty.
@@ -3863,7 +3863,7 @@ public S3AFileStatus probePathStatus(final Path path,
     @Override
     public RemoteIterator<S3ALocatedFileStatus> listFilesIterator(final Path path,
         final boolean recursive) throws IOException {
-      return S3AFileSystem.this.innerListFiles(path, recursive, Listing.ACCEPT_ALL_OBJECTS, null);
+      return S3AFileSystem.this.innerListFiles(path, recursive, Listing.ACCEPT_ALL_BUT_S3N, null);
     }
   }
 
@@ -5090,7 +5090,7 @@ public RemoteIterator<S3ALocatedFileStatus> listFilesAndEmptyDirectories(
     final Path path = qualify(f);
     return trackDurationAndSpan(INVOCATION_LIST_FILES, path, () ->
         innerListFiles(path, recursive,
-            Listing.ACCEPT_ALL_OBJECTS,
+            Listing.ACCEPT_ALL_BUT_S3N,
             null));
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestEMRFSCompatibility.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a;
+
+import org.assertj.core.api.Assertions;
+import org.junit.Test;
+
+import org.apache.hadoop.fs.Path;
+
+import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
+import static org.apache.hadoop.fs.s3a.Constants.S3N_FOLDER_SUFFIX;
+
+/**
+ * This test verifies that the EMRFS or legacy S3N filesystem compatibility with
+ * S3A works as expected.
+ */
+public class ITestEMRFSCompatibility extends AbstractS3ATestBase {
+  private static final String SRC_DIR = "src";
+  private static final String DEST_DIR = "dest";
+  private static final String SUBDIR = "subdir";
+
+  @Test
+  public void testFileSystemOperationWithS3NFolderMarker() throws Throwable {
+    S3AFileSystem fs = getFileSystem();
+    Path basePath = methodPath();
+    Path src = new Path(basePath, SRC_DIR);
+    Path dest = new Path(basePath, DEST_DIR);
+    Path subdir = new Path(src, SUBDIR);
+    Path folderMarker = new Path(subdir, S3N_FOLDER_SUFFIX);
+
+    // write an empty S3N folder marker object
+    touch(fs, folderMarker);
+
+    // verify initial state
+    Assertions.assertThat(fs.listStatus(subdir))
+        .describedAs("No objects are expected to be listed")
+        .isEmpty();
+
+    // rename the src folder.
+    fs.rename(src, dest);
+
+    //verify destination folder exists
+    Assertions.assertThat(fs.exists(new Path(dest, "subdir")))
+        .describedAs("Destination folder should exist")
+        .isTrue();
+
+    // verify src folder does not exists
+    Assertions.assertThat(fs.exists(src))
+        .describedAs("Source folder should not exist after rename")
+        .isFalse();
+  }
+}