Remove FileStatusAcceptor and Handle all cases for config values

Author: Bhavay Pahuja

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

@@ -1501,11 +1501,6 @@ private Constants() {
    */
   public static final String READ_RESTORED_GLACIER_OBJECTS = "fs.s3a.glacier.read.restored.objects";
 
-  /**
-   * Default value of Read Restored Glacier objects config.
-   */
-  public static final S3ObjectStorageClassFilter DEFAULT_READ_RESTORED_GLACIER_OBJECTS = S3ObjectStorageClassFilter.READ_ALL;
-
   /**
    * The bucket region header.
    */

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

@@ -18,7 +18,6 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import org.apache.hadoop.util.Lists;
 import software.amazon.awssdk.services.s3.model.CommonPrefix;
 import software.amazon.awssdk.services.s3.model.S3Object;
 
@@ -130,7 +129,7 @@ public static RemoteIterator<S3AFileStatus> toProvidedFileStatusIterator(
    * @param listPath path of the listing
    * @param request initial request to make
    * @param filter the filter on which paths to accept
-   * @param acceptors the class/predicate to decide which entries to accept
+   * @param acceptor the class/predicate to decide which entries to accept
    * in the listing based on the full file status.
    * @param span audit span for this iterator
    * @return the iterator
@@ -141,12 +140,12 @@ public FileStatusListingIterator createFileStatusListingIterator(
       Path listPath,
       S3ListRequest request,
       PathFilter filter,
-      List<FileStatusAcceptor> acceptors,
+      FileStatusAcceptor acceptor,
       AuditSpan span) throws IOException {
     return new FileStatusListingIterator(
         createObjectListingIterator(listPath, request, span),
         filter,
-        FileStatusAcceptor.resolveFileStatusAcceptors(acceptors)
+        acceptor
         );
   }
 
@@ -194,14 +193,14 @@ public RemoteIterator<S3ALocatedFileStatus> createSingleStatusIterator(
    * List files under a path assuming the path to be a directory.
    * @param path input path.
    * @param recursive recursive listing?
-   * @param acceptors file status filter
+   * @param acceptor file status filter
    * @param span audit span for this iterator
    * @return an iterator over listing.
    * @throws IOException any exception.
    */
   public RemoteIterator<S3ALocatedFileStatus> getListFilesAssumingDir(
       Path path,
-      boolean recursive, List<FileStatusAcceptor> acceptors,
+      boolean recursive, FileStatusAcceptor acceptor,
       AuditSpan span) throws IOException {
 
     String key = maybeAddTrailingSlash(pathToKey(path));
@@ -219,7 +218,7 @@ public RemoteIterator<S3ALocatedFileStatus> getListFilesAssumingDir(
                     delimiter,
                     span),
             ACCEPT_ALL,
-            acceptors,
+            acceptor,
             span));
   }
 
@@ -241,7 +240,7 @@ public RemoteIterator<S3ALocatedFileStatus> getLocatedFileStatusIteratorForDir(
             listingOperationCallbacks
                 .createListObjectsRequest(key, "/", span),
             filter,
-            Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter), new AcceptAllButSelfAndS3nDirs(dir)),
+            new AcceptAllButSelfAndS3nDirs(dir),
             span));
   }
 
@@ -270,8 +269,7 @@ public RemoteIterator<S3ALocatedFileStatus> getLocatedFileStatusIteratorForDir(
         path,
         request,
         ACCEPT_ALL,
-        Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter),
-            new AcceptAllButSelfAndS3nDirs(path)),
+        new AcceptAllButSelfAndS3nDirs(path),
         span);
   }
 
@@ -491,7 +489,8 @@ private boolean buildNextStatusBatch(S3ListResult objects) {
           LOG.debug("{}: {}", keyPath, stringify(s3Object));
         }
         // Skip over keys that are ourselves and old S3N _$folder$ files
-        if (acceptor.accept(keyPath, s3Object) && filter.accept(keyPath)) {
+        // Handle Glacier Storage Class objects based on the config fs.s3a.glacier.read.restored.objects value set
+        if ( s3ObjectStorageClassFilter.getFilter().apply(s3Object) &&  acceptor.accept(keyPath, s3Object) && filter.accept(keyPath)) {
           S3AFileStatus status = createFileStatus(keyPath, s3Object,
                   listingOperationCallbacks.getDefaultBlockSize(keyPath),
                   getStoreContext().getUsername(),
@@ -860,56 +859,6 @@ public boolean accept(FileStatus status) {
     }
   }
 
-
-  /**
-   * Accept all entries except unrestored Glacier objects if the config fs.s3a.glacier.read.restored.objects,
-   * is set to READ_RESTORED_GLACIER_OBJECTS
-   * Accept all entries except Glacier objects if the config fs.s3a.glacier.read.restored.objects,
-   * is set to SKIP_ALL_GLACIER
-   */
-  public static class GlacierStatusAcceptor implements FileStatusAcceptor {
-
-    private final S3ObjectStorageClassFilter s3ObjectStorageClassFilter;
-
-
-    public GlacierStatusAcceptor(S3ObjectStorageClassFilter s3ObjectStorageClassFilter) {
-      this.s3ObjectStorageClassFilter = s3ObjectStorageClassFilter;
-    }
-
-    /**
-     * Reject a s3Object entry with the Glacier storage class if the config fs.s3a.glacier.read.restored.objects is set to SKIP_ALL_GLACIER
-     * Reject a s3Object entry with the Glacier storage class if the config fs.s3a.glacier.read.restored.objects is set to READ_RESTORED_GLACIER_OBJECTS
-     * Accept a s3Object entry with the Glacier storage class if the config fs.s3a.glacier.read.restored.objects is set to READ_RESTORED_GLACIER_OBJECTS and the Glacier file is restored completely
-     * Reject a s3Object entry with any other storage class
-     * @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 s3ObjectStorageClassFilter.getFilter().apply(s3Object);
-    }
-
-    /**
-     * Accept all prefixes
-     * @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 true;
-    }
-
-    @Override
-    public boolean accept(FileStatus status) {
-      return true;
-    }
-
-  }
-
   @SuppressWarnings("unchecked")
   public static RemoteIterator<LocatedFileStatus> toLocatedFileStatusIterator(
       RemoteIterator<? extends LocatedFileStatus> iterator) {

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

@@ -52,8 +52,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import javax.annotation.Nullable;
 
-import org.apache.hadoop.fs.s3a.Listing.AcceptAllButSelfAndS3nDirs;
-import org.apache.hadoop.fs.s3a.Listing.GlacierStatusAcceptor;
 import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
@@ -147,6 +145,7 @@
 import org.apache.hadoop.fs.s3a.impl.StatusProbeEnum;
 import org.apache.hadoop.fs.s3a.impl.StoreContext;
 import org.apache.hadoop.fs.s3a.impl.StoreContextBuilder;
+import org.apache.hadoop.fs.s3a.Listing.AcceptAllButSelfAndS3nDirs;
 import org.apache.hadoop.fs.s3a.prefetch.S3APrefetchingInputStream;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperations;
 import org.apache.hadoop.fs.s3a.tools.MarkerToolOperationsImpl;
@@ -586,8 +585,10 @@ public void initialize(URI name, Configuration originalConf)
       s3aInternals = createS3AInternals();
 
       s3ObjectStorageClassFilter = Optional.ofNullable(conf.get(READ_RESTORED_GLACIER_OBJECTS))
+          .map(String::trim)
+          .map(String::toUpperCase)
           .map(S3ObjectStorageClassFilter::valueOf)
-          .orElse(DEFAULT_READ_RESTORED_GLACIER_OBJECTS);
+          .orElse(S3ObjectStorageClassFilter.READ_ALL);
 
       // look for encryption data
       // DT Bindings may override this
@@ -2484,8 +2485,8 @@ public RemoteIterator<S3ALocatedFileStatus> listFilesAndDirectoryMarkers(
           path,
           true,
           includeSelf
-              ? Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter), Listing.ACCEPT_ALL_BUT_S3N)
-              : Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter), new AcceptAllButSelfAndS3nDirs(path)),
+              ? Listing.ACCEPT_ALL_BUT_S3N
+              : new AcceptAllButSelfAndS3nDirs(path),
           status
       );
     }
@@ -2533,7 +2534,7 @@ public RemoteIterator<S3AFileStatus> listObjects(
           listing.createFileStatusListingIterator(path,
               createListObjectsRequest(key, null),
               ACCEPT_ALL,
-              Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter), Listing.ACCEPT_ALL_BUT_S3N),
+              Listing.ACCEPT_ALL_BUT_S3N,
               auditSpan));
     }
 
@@ -3870,8 +3871,7 @@ public S3AFileStatus probePathStatus(final Path path,
     public RemoteIterator<S3ALocatedFileStatus> listFilesIterator(final Path path,
         final boolean recursive) throws IOException {
       return S3AFileSystem.this.innerListFiles(path, recursive,
-          Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter),
-              Listing.ACCEPT_ALL_BUT_S3N), null);
+              Listing.ACCEPT_ALL_BUT_S3N, null);
     }
   }
 
@@ -5174,7 +5174,7 @@ public RemoteIterator<LocatedFileStatus> listFiles(Path f,
     return toLocatedFileStatusIterator(
         trackDurationAndSpan(INVOCATION_LIST_FILES, path, () ->
             innerListFiles(path, recursive,
-                Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter), new Listing.AcceptFilesOnly(path)), null)));
+                new Listing.AcceptFilesOnly(path), null)));
   }
 
   /**
@@ -5192,7 +5192,7 @@ public RemoteIterator<S3ALocatedFileStatus> listFilesAndEmptyDirectories(
     final Path path = qualify(f);
     return trackDurationAndSpan(INVOCATION_LIST_FILES, path, () ->
         innerListFiles(path, recursive,
-            Lists.newArrayList(new GlacierStatusAcceptor(s3ObjectStorageClassFilter), Listing.ACCEPT_ALL_BUT_S3N),
+            Listing.ACCEPT_ALL_BUT_S3N,
             null));
   }
 
@@ -5207,7 +5207,7 @@ public RemoteIterator<S3ALocatedFileStatus> listFilesAndEmptyDirectories(
    *
    * @param f path
    * @param recursive recursive listing?
-   * @param acceptors file status filters
+   * @param acceptor file status filters
    * @param status optional status of path to list.
    * @return an iterator over the listing.
    * @throws IOException failure
@@ -5216,7 +5216,7 @@ public RemoteIterator<S3ALocatedFileStatus> listFilesAndEmptyDirectories(
   private RemoteIterator<S3ALocatedFileStatus> innerListFiles(
       final Path f,
       final boolean recursive,
-      final List<Listing.FileStatusAcceptor> acceptors,
+      final Listing.FileStatusAcceptor acceptor,
       final S3AFileStatus status) throws IOException {
     Path path = qualify(f);
     LOG.debug("listFiles({}, {})", path, recursive);
@@ -5233,8 +5233,8 @@ private RemoteIterator<S3ALocatedFileStatus> innerListFiles(
       RemoteIterator<S3ALocatedFileStatus> listFilesAssumingDir =
               listing.getListFilesAssumingDir(path,
                       recursive,
-                  acceptors,
-                  getActiveAuditSpan());
+                      acceptor,
+                      getActiveAuditSpan());
       // If there are no list entries present, we
       // fallback to file existence check as the path
       // can be a file or empty directory.