HADOOP-15798. LocalMetadataStore put() does not retain isDeleted in parent listing. Contributed by Gabor Bota.

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

@@ -31,6 +31,9 @@ public final class LocalMetadataEntry {
   @Nullable
   private DirListingMetadata dirListingMetadata;
 
+  LocalMetadataEntry() {
+  }
+
   LocalMetadataEntry(PathMetadata pmd){
     pathMetadata = pmd;
     dirListingMetadata = null;

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

@@ -269,19 +269,28 @@ public void put(PathMetadata meta) throws IOException {
       Path parentPath = path.getParent();
       if (parentPath != null) {
         LocalMetadataEntry parentMeta = localCache.getIfPresent(parentPath);
+
+        // Create empty parent LocalMetadataEntry if it doesn't exist
+        if (parentMeta == null){
+          parentMeta = new LocalMetadataEntry();
+          localCache.put(parentPath, parentMeta);
+        }
+
+        // If there is no directory metadata on the parent entry, create
+        // an empty one
+        if (!parentMeta.hasDirMeta()) {
           DirListingMetadata parentDirMeta =
               new DirListingMetadata(parentPath, DirListingMetadata.EMPTY_DIR,
                   false);
-        parentDirMeta.put(status);
-
-        getDirListingMeta(parentPath);
-
-        if (parentMeta == null){
-          localCache.put(parentPath, new LocalMetadataEntry(parentDirMeta));
-        } else if (!parentMeta.hasDirMeta()) {
           parentMeta.setDirListingMetadata(parentDirMeta);
-        } else {
+        }
+
+        // Add the child status to the listing
         parentMeta.getDirListingMeta().put(status);
+
+        // Mark the listing entry as deleted if the meta is set to deleted
+        if(meta.isDeleted()) {
+          parentMeta.getDirListingMeta().markDeleted(path);
         }
       }
     }

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

@@ -178,7 +178,8 @@ public void testRollingRenames() throws Exception {
   public void testConsistentListAfterDelete() throws Exception {
     S3AFileSystem fs = getFileSystem();
     // test will fail if NullMetadataStore (the default) is configured: skip it.
-    Assume.assumeTrue(fs.hasMetadataStore());
+    Assume.assumeTrue("FS needs to have a metadatastore.",
+        fs.hasMetadataStore());
 
     // Any S3 keys that contain DELAY_KEY_SUBSTRING will be delayed
     // in listObjects() results via InconsistentS3Client
@@ -190,11 +191,11 @@ public void testConsistentListAfterDelete() throws Exception {
         inconsistentPath};
 
     for (Path path : testDirs) {
-      assertTrue(fs.mkdirs(path));
+      assertTrue("Can't create directory: " + path, fs.mkdirs(path));
     }
     clearInconsistency(fs);
     for (Path path : testDirs) {
-      assertTrue(fs.delete(path, false));
+      assertTrue("Can't delete path: " + path, fs.delete(path, false));
     }
 
     FileStatus[] paths = fs.listStatus(path("a/b/"));
@@ -202,10 +203,13 @@ public void testConsistentListAfterDelete() throws Exception {
     for (FileStatus fileState : paths) {
       list.add(fileState.getPath());
     }
-    assertFalse(list.contains(path("a/b/dir1")));
+
-    assertFalse(list.contains(path("a/b/dir2")));
+    assertFalse("This path should be deleted.",
-    // This should fail without S3Guard, and succeed with it.
+        list.contains(path("a/b/dir1")));
-    assertFalse(list.contains(inconsistentPath));
+    assertFalse("This path should be deleted.",
+        list.contains(path("a/b/dir2")));
+    assertFalse("This should fail without S3Guard, and succeed with it.",
+        list.contains(inconsistentPath));
   }
 
   /**

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

@@ -808,6 +808,21 @@ public void testPutDirListingMetadataPutsFileMetadata()
     }
   }
 
+  @Test
+  public void testPutRetainsIsDeletedInParentListing() throws Exception {
+    final Path path = strToPath("/a/b");
+    final FileStatus fileStatus = basicFileStatus(path, 0, false);
+    PathMetadata pm = new PathMetadata(fileStatus);
+    pm.setIsDeleted(true);
+    ms.put(pm);
+    if(!allowMissing()) {
+      final PathMetadata pathMetadata =
+          ms.listChildren(path.getParent()).get(path);
+      assertTrue("isDeleted should be true on the parent listing",
+          pathMetadata.isDeleted());
+    }
+  }
+
   /*
    * Helper functions.
    */