HADOOP-16430. S3AFilesystem.delete to incrementally update s3guard with deletions

Contributed by Steve Loughran.

This overlaps the scanning for directory entries with batched calls to S3 DELETE and updates of the S3Guard tables.
It also uses S3Guard to list the files to delete, so find newly created files even when S3 listings are not use consistent.

For path which the client considers S3Guard to be authoritative, we also do a recursive LIST of the store and delete files; this is to find unindexed files and do guarantee that the delete(path, true) call really does delete everything underneath.

Change-Id: Ice2f6e940c506e0b3a78fa534a99721b1698708e

Author: steveloughran

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/impl/FutureIOSupport.java

@@ -144,13 +144,13 @@ private static IOException unwrapInnerException(final Throwable e) {
     Throwable cause = e.getCause();
     if (cause instanceof IOException) {
       return (IOException) cause;
-    } else if (cause instanceof WrappedIOException){
+    } else if (cause instanceof WrappedIOException) {
       return ((WrappedIOException) cause).getCause();
-    } else if (cause instanceof CompletionException){
+    } else if (cause instanceof CompletionException) {
       return unwrapInnerException(cause);
-    } else if (cause instanceof ExecutionException){
+    } else if (cause instanceof ExecutionException) {
       return unwrapInnerException(cause);
-    } else if (cause instanceof RuntimeException){
+    } else if (cause instanceof RuntimeException) {
       throw (RuntimeException) cause;
     } else if (cause != null) {
       // other type: wrap with a new IOE

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextURIBase.java

@@ -130,7 +130,7 @@ public void testCreateFileWithNullName() throws IOException {
 
   @Test
   public void testCreateExistingFile() throws IOException {
-    String fileName = "testFile";
+    String fileName = "testCreateExistingFile";
     Path testPath = qualifiedPath(fileName, fc2);
 
     // Ensure file does not exist
@@ -153,7 +153,7 @@ public void testCreateExistingFile() throws IOException {
 
   @Test
   public void testCreateFileInNonExistingDirectory() throws IOException {
-    String fileName = "testDir/testFile";
+    String fileName = "testCreateFileInNonExistingDirectory/testFile";
 
     Path testPath = qualifiedPath(fileName, fc2);
 
@@ -165,7 +165,8 @@ public void testCreateFileInNonExistingDirectory() throws IOException {
 
     // Ensure using fc2 that file is created
     Assert.assertTrue(isDir(fc2, testPath.getParent()));
-    Assert.assertEquals("testDir", testPath.getParent().getName());
+    Assert.assertEquals("testCreateFileInNonExistingDirectory",
+        testPath.getParent().getName());
     Assert.assertTrue(exists(fc2, testPath));
 
   }
@@ -293,7 +294,7 @@ public void testIsDirectory() throws IOException {
 
   @Test
   public void testDeleteFile() throws IOException {
-    Path testPath = qualifiedPath("testFile", fc2);
+    Path testPath = qualifiedPath("testDeleteFile", fc2);
 
     // Ensure file does not exist
     Assert.assertFalse(exists(fc2, testPath));
@@ -314,7 +315,7 @@ public void testDeleteFile() throws IOException {
 
   @Test
   public void testDeleteNonExistingFile() throws IOException {
-    String testFileName = "testFile";
+    String testFileName = "testDeleteNonExistingFile";
     Path testPath = qualifiedPath(testFileName, fc2);
 
     // TestCase1 : Test delete on file never existed
@@ -341,7 +342,7 @@ public void testDeleteNonExistingFile() throws IOException {
 
   @Test
   public void testDeleteNonExistingFileInDir() throws IOException {
-    String testFileInDir = "testDir/testDir/TestFile";
+    String testFileInDir = "testDeleteNonExistingFileInDir/testDir/TestFile";
     Path testPath = qualifiedPath(testFileInDir, fc2);
 
     // TestCase1 : Test delete on file never existed
@@ -418,7 +419,7 @@ public void testDeleteDirectory() throws IOException {
 
   @Test
   public void testDeleteNonExistingDirectory() throws IOException {
-    String testDirName = "testFile";
+    String testDirName = "testDeleteNonExistingDirectory";
     Path testPath = qualifiedPath(testDirName, fc2);
 
     // TestCase1 : Test delete on directory never existed
@@ -445,7 +446,7 @@ public void testDeleteNonExistingDirectory() throws IOException {
 
   @Test
   public void testModificationTime() throws IOException {
-    String testFile = "file1";
+    String testFile = "testModificationTime";
     long fc2ModificationTime, fc1ModificationTime;
 
     Path testPath = qualifiedPath(testFile, fc2);
@@ -461,7 +462,7 @@ public void testModificationTime() throws IOException {
 
   @Test
   public void testFileStatus() throws IOException {
-    String fileName = "file1";
+    String fileName = "testModificationTime";
     Path path2 = fc2.makeQualified(new Path(BASE, fileName));
 
     // Create a file on fc2's file system using fc1

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/AbstractContractGetFileStatusTest.java

@@ -112,9 +112,11 @@ public void testListFilesEmptyDirectoryRecursive() throws IOException {
   private void listFilesOnEmptyDir(boolean recursive) throws IOException {
     describe("Invoke listFiles(recursive=" + recursive + ")" +
         " on empty directories, expect nothing found");
-    Path subfolder = createDirWithEmptySubFolder();
     FileSystem fs = getFileSystem();
-    new TreeScanResults(fs.listFiles(getContract().getTestPath(), recursive))
+    Path path = getContract().getTestPath();
+    fs.delete(path, true);
+    Path subfolder = createDirWithEmptySubFolder();
+    new TreeScanResults(fs.listFiles(path, recursive))
         .assertSizeEquals("listFiles(test dir, " + recursive + ")", 0, 0, 0);
     describe("Test on empty subdirectory");
     new TreeScanResults(fs.listFiles(subfolder, recursive))
@@ -126,9 +128,11 @@ private void listFilesOnEmptyDir(boolean recursive) throws IOException {
   public void testListLocatedStatusEmptyDirectory() throws IOException {
     describe("Invoke listLocatedStatus() on empty directories;" +
         " expect directories to be found");
-    Path subfolder = createDirWithEmptySubFolder();
     FileSystem fs = getFileSystem();
-    new TreeScanResults(fs.listLocatedStatus(getContract().getTestPath()))
+    Path path = getContract().getTestPath();
+    fs.delete(path, true);
+    Path subfolder = createDirWithEmptySubFolder();
+    new TreeScanResults(fs.listLocatedStatus(path))
       .assertSizeEquals("listLocatedStatus(test dir)", 0, 1, 0);
     describe("Test on empty subdirectory");
     new TreeScanResults(fs.listLocatedStatus(subfolder))

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java

@@ -1627,6 +1627,22 @@ public String toString() {
           getFileCount() == 1 ? "" : "s");
     }
 
+    /**
+     * Dump the files and directories to a multi-line string for error
+     * messages and assertions.
+     * @return a dump of the internal state
+     */
+    private String dump() {
+      StringBuilder sb = new StringBuilder(toString());
+      sb.append("\nFiles:");
+      directories.forEach(p ->
+          sb.append("\n  \"").append(p.toString()));
+      sb.append("\nDirectories:");
+      files.forEach(p ->
+          sb.append("\n  \"").append(p.toString()));
+      return sb.toString();
+    }
+
     /**
      * Equality check compares files and directory counts.
      * As these are non-final fields, this class cannot be used in
@@ -1667,7 +1683,7 @@ public int hashCode() {
      * @param o expected other entries.
      */
     public void assertSizeEquals(String text, long f, long d, long o) {
-      String self = toString();
+      String self = dump();
       Assert.assertEquals(text + ": file count in " + self,
           f, getFileCount());
       Assert.assertEquals(text + ": directory count in " + self,

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

@@ -199,6 +199,7 @@ public DeleteObjectsResult deleteObjects(DeleteObjectsRequest
       deleteObjectsRequest)
       throws AmazonClientException, AmazonServiceException {
     maybeFail();
+    LOG.info("registering bulk delete of objects");
     for (DeleteObjectsRequest.KeyVersion keyVersion :
         deleteObjectsRequest.getKeys()) {
       registerDeleteObject(keyVersion.getKey(),
@@ -278,6 +279,7 @@ private void addSummaryIfNotPresent(List<S3ObjectSummary> list,
     // Behavior of S3ObjectSummary
     String key = item.getKey();
     if (list.stream().noneMatch((member) -> member.getKey().equals(key))) {
+      LOG.debug("Reinstate summary {}", key);
       list.add(item);
     }
   }
@@ -302,6 +304,7 @@ private void addPrefixIfNotPresent(List<String> prefixes, String ancestor,
       if (nextParent.equals(ancestorPath)) {
         String prefix = prefixCandidate.toString();
         if (!prefixes.contains(prefix)) {
+          LOG.debug("Reinstate prefix {}", prefix);
           prefixes.add(prefix);
         }
         return;
@@ -401,6 +404,7 @@ private void restoreDeleted(List<S3ObjectSummary> summaries,
         }
       } else {
         // Clean up any expired entries
+        LOG.debug("Remove expired key {}", key);
         delayedDeletes.remove(key);
       }
     }
@@ -467,16 +471,24 @@ private boolean isKeyDelayed(Long enqueueTime, String key) {
 
   private void registerDeleteObject(String key, String bucket) {
     if (policy.shouldDelay(key)) {
-      // Record summary so we can add it back for some time post-deletion
-      ListObjectsRequest request = new ListObjectsRequest()
-              .withBucketName(bucket)
-              .withPrefix(key);
-      S3ObjectSummary summary = innerlistObjects(request).getObjectSummaries()
-          .stream()
-          .filter(result -> result.getKey().equals(key))
-          .findFirst()
-          .orElse(null);
-      delayedDeletes.put(key, new Delete(System.currentTimeMillis(), summary));
+      Delete delete = delayedDeletes.get(key);
+      if (delete != null && isKeyDelayed(delete.time(), key)) {
+        // there is already an entry in the delayed delete list,
+        // so ignore the operation
+        LOG.debug("Ignoring delete of already deleted object");
+      } else {
+        // Record summary so we can add it back for some time post-deletion
+        ListObjectsRequest request = new ListObjectsRequest()
+            .withBucketName(bucket)
+            .withPrefix(key);
+        S3ObjectSummary summary = innerlistObjects(request).getObjectSummaries()
+            .stream()
+            .filter(result -> result.getKey().equals(key))
+            .findFirst()
+            .orElse(null);
+        delayedDeletes.put(key, new Delete(System.currentTimeMillis(),
+            summary));
+      }
     }
   }
 

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

@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.fs.s3a;
 
+import javax.annotation.Nullable;
+
 import com.amazonaws.AmazonClientException;
 import com.amazonaws.services.s3.model.S3ObjectSummary;
 import com.google.common.annotations.VisibleForTesting;
@@ -58,6 +60,9 @@ public class Listing {
   private final S3AFileSystem owner;
   private static final Logger LOG = S3AFileSystem.LOG;
 
+  static final FileStatusAcceptor ACCEPT_ALL_BUT_S3N =
+      new AcceptAllButS3nDirs();
+
   public Listing(S3AFileSystem owner) {
     this.owner = owner;
   }
@@ -339,7 +344,8 @@ class FileStatusListingIterator
     FileStatusListingIterator(ObjectListingIterator source,
         PathFilter filter,
         FileStatusAcceptor acceptor,
-        RemoteIterator<S3AFileStatus> providedStatus) throws IOException {
+        @Nullable RemoteIterator<S3AFileStatus> providedStatus)
+        throws IOException {
       this.source = source;
       this.filter = filter;
       this.acceptor = acceptor;

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

@@ -31,6 +31,9 @@
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
 public class S3AFileStatus extends FileStatus {
+
+  private static final long serialVersionUID = -5955674081978903922L;
+
   private Tristate isEmptyDirectory;
   private String eTag;
   private String versionId;
@@ -56,12 +59,16 @@ public S3AFileStatus(boolean isemptydir,
   public S3AFileStatus(Tristate isemptydir,
       Path path,
       String owner) {
-    super(0, true, 1, 0, 0, 0,
-        null, null, null, null,
-        path, false, true, false);
-    isEmptyDirectory = isemptydir;
-    setOwner(owner);
-    setGroup(owner);
+    this(path,
+        true,
+        isemptydir,
+        0,
+        0,
+        0,
+        owner,
+        null,
+        null
+    );
   }
 
   /**
@@ -76,10 +83,43 @@ public S3AFileStatus(Tristate isemptydir,
    */
   public S3AFileStatus(long length, long modification_time, Path path,
       long blockSize, String owner, String eTag, String versionId) {
-    super(length, false, 1, blockSize, modification_time,
+    this(path,
+        false,
+        Tristate.FALSE,
+        length,
+        modification_time,
+        blockSize,
+        owner,
+        eTag,
+        versionId
+    );
+  }
+
+  /**
+   * Either a file or directory.
+   * @param path path
+   * @param isDir is this a directory?
+   * @param isemptydir is this an empty directory?
+   * @param length file length
+   * @param modificationTime mod time
+   * @param blockSize block size
+   * @param owner owner
+   * @param eTag eTag of the S3 object if available, else null
+   * @param versionId versionId of the S3 object if available, else null
+   */
+  S3AFileStatus(Path path,
+      boolean isDir,
+      Tristate isemptydir,
+      long length,
+      long modificationTime,
+      long blockSize,
+      String owner,
+      String eTag,
+      String versionId) {
+    super(length, isDir, 1, blockSize, modificationTime,
         0, null, owner, owner, null,
         path, false, true, false);
-    isEmptyDirectory = Tristate.FALSE;
+    this.isEmptyDirectory = isemptydir;
     this.eTag = eTag;
     this.versionId = versionId;
   }