fixes for issues found during testing with localmetadatastore

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

@@ -550,6 +550,8 @@ public static void putWithTtl(MetadataStore ms, DirListingMetadata dirMeta,
       ITtlTimeProvider timeProvider)
       throws IOException {
     dirMeta.setLastUpdated(timeProvider.getNow());
+    dirMeta.getListing()
+        .forEach(pm -> pm.setLastUpdated(timeProvider.getNow()));
     ms.put(dirMeta);
   }
 
@@ -622,8 +624,6 @@ public static DirListingMetadata listChildrenWithTtl(MetadataStore ms,
 
     long ttl = timeProvider.getMetadataTtl();
 
-
-
     if (dlm != null && dlm.isAuthoritative()
         && dlm.isExpired(ttl, timeProvider.getNow())) {
       dlm.setAuthoritative(false);

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

@@ -27,10 +27,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 
-import org.apache.hadoop.fs.LocatedFileStatus;
-import org.apache.hadoop.fs.RemoteIterator;
 import org.junit.Assume;
-import org.apache.hadoop.fs.s3a.s3guard.ITtlTimeProvider;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -49,7 +46,11 @@
 import org.apache.hadoop.fs.s3a.s3guard.DirListingMetadata;
 import org.apache.hadoop.fs.s3a.s3guard.MetadataStore;
 import org.apache.hadoop.fs.s3a.s3guard.PathMetadata;
+import org.apache.hadoop.fs.s3a.s3guard.ITtlTimeProvider;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.RemoteIterator;
 
+import static org.apache.hadoop.fs.contract.ContractTestUtils.touch;
 import static org.apache.hadoop.fs.s3a.S3ATestUtils.removeBaseAndBucketOverrides;
 import static org.apache.hadoop.test.LambdaTestUtils.eventually;
 import static org.junit.Assume.assumeTrue;
@@ -383,7 +384,7 @@ private void deleteGuardedTombstoned(S3AFileSystem guardedFs,
    * created out of band.
    */
   @Test
-  public void createNonRecursiveFailsIfParentDeleted() throws Exception {
+  public void testCreateNonRecursiveFailsIfParentDeleted() throws Exception {
     LOG.info("Authoritative mode: {}", authoritative);
 
     String dirToDelete = methodName + UUID.randomUUID().toString();
@@ -430,6 +431,54 @@ public void createNonRecursiveFailsIfParentDeleted() throws Exception {
     }
   }
 
+  /**
+   * When lastUpdated = 0 the entry should not expire. This is a special case
+   * eg. for old metadata entries
+   */
+  @Test
+  public void testLastUpdatedZeroWontExpire() throws Exception {
+    LOG.info("Authoritative mode: {}", authoritative);
+
+    String testFile = methodName + UUID.randomUUID().toString() +
+        "/theFileToTry";
+
+    long ttl = 10L;
+    final Path filePath = path(testFile);
+
+    // Create a directory with
+    ITtlTimeProvider mockTimeProvider = mock(ITtlTimeProvider.class);
+    ITtlTimeProvider originalTimeProvider = guardedFs.getTtlTimeProvider();
+
+    try {
+      guardedFs.setTtlTimeProvider(mockTimeProvider);
+      when(mockTimeProvider.getMetadataTtl()).thenReturn(ttl);
+
+      // create a file while the NOW is 0, so it will set 0 as the last_upadted
+      when(mockTimeProvider.getNow()).thenReturn(0L);
+      touch(guardedFs, filePath);
+      deleteFile(guardedFs, filePath);
+
+      final PathMetadata pathMetadata =
+          guardedFs.getMetadataStore().get(filePath);
+      assertNotNull("pathMetadata should not be null after deleting with "
+          + "tombstones", pathMetadata);
+      assertEquals("pathMetadata lastUpdated field should be 0", 0,
+          pathMetadata.getLastUpdated());
+
+      // set the time, so the metadata would expire
+      when(mockTimeProvider.getNow()).thenReturn(2*ttl);
+      intercept(FileNotFoundException.class, filePath.toString(),
+          "This file should throw FNFE when reading through "
+              + "the guarded fs, and the metadatastore tombstoned the file. "
+              + "The tombstone won't expire if lastUpdated is set to 0.",
+          () -> guardedFs.getFileStatus(filePath));
+
+    } finally {
+      guardedFs.delete(filePath, true);
+      guardedFs.setTtlTimeProvider(originalTimeProvider);
+    }
+  }
+
   private void checkListingDoesNotContainPath(S3AFileSystem fs, Path filePath)
       throws IOException {
     final RemoteIterator<LocatedFileStatus> listIter =

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

@@ -206,7 +206,7 @@ public void testFileMetadataExpiresTtl() throws Exception {
    * create(tombstone file) must succeed irrespective of overwrite flag.
    */
   @Test
-  public void createOnTombstonedFileSucceeds() throws Exception {
+  public void testCreateOnTombstonedFileSucceeds() throws Exception {
     LOG.info("Authoritative mode: {}", authoritative);
     final S3AFileSystem fs = getFileSystem();
 
@@ -244,7 +244,7 @@ public void createOnTombstonedFileSucceeds() throws Exception {
    * must be valid. That is: the putAncestor code will correct everything
    */
   @Test
-  public void createParentHasTombstone() throws Exception {
+  public void testCreateParentHasTombstone() throws Exception {
     LOG.info("Authoritative mode: {}", authoritative);
     final S3AFileSystem fs = getFileSystem();