HADOOP-16644. Do a HEAD after a PUT to get the modtime.

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

@@ -3420,9 +3420,32 @@ void finishedWrite(String key, long length, String eTag, String versionId,
           activeState = stateToClose;
         }
         S3Guard.addAncestors(metadataStore, p, ttlTimeProvider, activeState);
-        S3AFileStatus status = createUploadFileStatus(p,
-            S3AUtils.objectRepresentsDirectory(key, length), length,
-            getDefaultBlockSize(p), username, eTag, versionId);
+        boolean isDir = objectRepresentsDirectory(key, length);
+        S3AFileStatus status = null;
+        status = createUploadFileStatus(p,
+            isDir, length,
+            getDefaultBlockSize(p), username, , versionId);
+        // do a HEAD to pick up the real timestamp. This is a PITA but
+        // it is the only way to defend against clock drift and timestamp
+        // inconsistencies, which can cause surprises later.
+        S3AFileStatus remoteStatus = null;
+        if (!isDir && eTag != null) {
+          // we need to pass down the version ID/etag so on an update we
+          // can discard previous results
+          try {
+            remoteStatus = innerGetFileStatus(p, false,
+                StatusProbeEnum.HEAD_ONLY);
+            if (eTag.equals(remoteStatus.getETag())) {
+              status = new S3AFileStatus(status.getLen(),
+                  remoteStatus.getModificationTime(),
+                  status.getPath(), status.getBlockSize(), status.getOwner(),
+                  eTag, status.getVersionId());
+            }
+          } catch (IOException ignored) {
+            // we don't worry if the file isn't visible or some other
+            // failure occurs.
+          }
+        }
         S3Guard.putAndReturn(metadataStore, status,
             instrumentation,
             ttlTimeProvider,

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

@@ -41,4 +41,16 @@ public enum StatusProbeEnum {
   public static final Set<StatusProbeEnum> DIRECTORIES =
       EnumSet.of(DirMarker, List);
 
+  /** We only want the HEAD or dir marker. */
+  public static final Set<StatusProbeEnum> HEAD_OR_DIR_MARKER =
+      EnumSet.of(Head, DirMarker);
+
+  /** We only want the HEAD. */
+  public static final Set<StatusProbeEnum> HEAD_ONLY =
+      EnumSet.of(Head);
+
+  /** We only want the dir marker. */
+  public static final Set<StatusProbeEnum> DIR_MARKER_ONLY =
+      EnumSet.of(DirMarker);
+
 }