Load content hash as the etag of the object when the UFS is S3

### What changes are proposed in this pull request?

Recode the content hash into XAttr as the ETag when the UFS is S3. When the file is newly created or loaded from UFS, Alluxio will record the ContentHash into metastore so the user can see the ETag attribute.

### Why are the changes needed?
When the UFS is Object storage, The user can see the ETag of the file persisted in UFS by this way.

### Does this PR introduce any user facing changes?
No

			pr-link: #18440
			change-id: cid-1204732240a5e73959b917eb6d9f0c97e05820dc

core/client/fs/src/main/java/alluxio/client/file/ufs/UfsBaseFileSystem.java

@@ -12,6 +12,7 @@
 package alluxio.client.file.ufs;
 
 import alluxio.AlluxioURI;
+import alluxio.Constants;
 import alluxio.client.file.FileInStream;
 import alluxio.client.file.FileOutStream;
 import alluxio.client.file.FileSystem;
@@ -80,6 +81,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -476,6 +478,10 @@ private URIStatus transformStatus(UfsStatus ufsStatus) {
       info.setUfsFingerprint(
           Fingerprint.create(mUfs.get().getUnderFSType(), ufsStatus, fileStatus.getContentHash())
                      .serialize());
+      if (info.getXAttr() == null) {
+        info.setXAttr(new HashMap<String, byte[]>());
+      }
+      info.getXAttr().put(Constants.ETAG_XATTR_KEY, fileStatus.getContentHash().getBytes());
     }
     else {
       info.setLength(0);

core/common/src/main/java/alluxio/Constants.java

@@ -237,5 +237,8 @@ public final class Constants {
   public static final Pattern LOG_FILE_PATTERN =
       Pattern.compile(".*(\\.log|\\.out)(\\.[0-9-]+)?$|.*.txt|.*.json");
 
+  /* xAttr keys */
+  public static final String ETAG_XATTR_KEY = "s3_etag";
+
   private Constants() {} // prevent instantiation
 }

core/server/master/src/main/java/alluxio/master/file/DefaultFileSystemMaster.java

@@ -205,13 +205,15 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
 import com.google.common.collect.Streams;
+import com.google.protobuf.ByteString;
 import io.grpc.ServerInterceptors;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.text.MessageFormat;
 import java.time.Clock;
 import java.util.ArrayList;
@@ -1806,14 +1808,15 @@ void completeFileInternal(RpcContext rpcContext, LockedInodePath inodePath,
     long length = fileInode.isPersisted() ? context.getOptions().getUfsLength() : inAlluxioLength;
 
     String ufsFingerprint = Constants.INVALID_UFS_FINGERPRINT;
+    String contentHash = null;
     if (fileInode.isPersisted()) {
-      String contentHash = context.getOptions().hasContentHash()
+      contentHash = context.getOptions().hasContentHash()
           ? context.getOptions().getContentHash() : null;
       ufsFingerprint = getUfsFingerprint(inodePath.getUri(), context.getUfsStatus(), contentHash);
     }
 
     completeFileInternal(rpcContext, inodePath, length, context.getOperationTimeMs(),
-        ufsFingerprint);
+        ufsFingerprint, contentHash);
   }
 
   /**
@@ -1824,7 +1827,7 @@ void completeFileInternal(RpcContext rpcContext, LockedInodePath inodePath,
    * @param ufsFingerprint the ufs fingerprint
    */
   private void completeFileInternal(RpcContext rpcContext, LockedInodePath inodePath, long length,
-      long opTimeMs, String ufsFingerprint)
+      long opTimeMs, String ufsFingerprint, String contentHash)
       throws FileDoesNotExistException, InvalidPathException, InvalidFileSizeException,
       FileAlreadyCompletedException, UnavailableException {
     Preconditions.checkState(inodePath.getLockPattern().isWrite());
@@ -1865,15 +1868,19 @@ private void completeFileInternal(RpcContext rpcContext, LockedInodePath inodePa
       mUfsAbsentPathCache.processExisting(inodePath.getUri());
     }
 
-    // We could introduce a concept of composite entries, so that these two entries could
-    // be applied in a single call to applyAndJournal.
-    mInodeTree.updateInode(rpcContext, UpdateInodeEntry.newBuilder()
-        .setId(inode.getId())
-        .setUfsFingerprint(ufsFingerprint)
+    UpdateInodeEntry.Builder updateEntry = UpdateInodeEntry.newBuilder().setId(inode.getId());
+    updateEntry.setUfsFingerprint(ufsFingerprint)
         .setLastModificationTimeMs(opTimeMs)
         .setLastAccessTimeMs(opTimeMs)
-        .setOverwriteModificationTime(true)
-        .build());
+        .setOverwriteModificationTime(true);
+    if (StringUtils.isNotEmpty(contentHash)) {
+      updateEntry.putXAttr(Constants.ETAG_XATTR_KEY,
+              ByteString.copyFrom(contentHash, StandardCharsets.UTF_8))
+          .setXAttrUpdateStrategy(File.XAttrUpdateStrategy.UNION_REPLACE);
+    }
+    // We could introduce a concept of composite entries, so that these two entries could
+    // be applied in a single call to applyAndJournal.
+    mInodeTree.updateInode(rpcContext, updateEntry.build());
     mInodeTree.updateInodeFile(rpcContext, entry.build());
 
     Metrics.FILES_COMPLETED.inc();

core/server/master/src/main/java/alluxio/master/file/InodeSyncStream.java

@@ -12,6 +12,7 @@
 package alluxio.master.file;
 
 import alluxio.AlluxioURI;
+import alluxio.Constants;
 import alluxio.client.WriteType;
 import alluxio.collections.Pair;
 import alluxio.conf.Configuration;
@@ -1216,6 +1217,12 @@ void loadFileMetadataInternal(RpcContext rpcContext, LockedInodePath inodePath,
     createFileContext.setOwner(context.getUfsStatus().getOwner());
     createFileContext.setGroup(context.getUfsStatus().getGroup());
     createFileContext.setXAttr(context.getUfsStatus().getXAttr());
+    if (createFileContext.getXAttr() == null) {
+      createFileContext.setXAttr(new HashMap<String, byte[]>());
+    }
+    createFileContext.getXAttr().put(Constants.ETAG_XATTR_KEY,
+        ((UfsFileStatus) context.getUfsStatus()).getContentHash().getBytes());
+
     short ufsMode = context.getUfsStatus().getMode();
     Mode mode = new Mode(ufsMode);
     Long ufsLastModified = context.getUfsStatus().getLastModifiedTime();

core/server/master/src/main/java/alluxio/master/file/mdsync/DefaultSyncProcess.java

@@ -12,6 +12,7 @@
 package alluxio.master.file.mdsync;
 
 import alluxio.AlluxioURI;
+import alluxio.Constants;
 import alluxio.client.WriteType;
 import alluxio.collections.Pair;
 import alluxio.conf.Configuration;
@@ -77,6 +78,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Optional;
@@ -775,6 +777,11 @@ private List<Inode> createInodeFileMetadata(
     createFileContext.setOwner(ufsStatus.getOwner());
     createFileContext.setGroup(ufsStatus.getGroup());
     createFileContext.setXAttr(ufsStatus.getXAttr());
+    if (createFileContext.getXAttr() == null) {
+      createFileContext.setXAttr(new HashMap<String, byte[]>());
+    }
+    createFileContext.getXAttr()
+        .put(Constants.ETAG_XATTR_KEY, ((UfsFileStatus) ufsStatus).getContentHash().getBytes());
     short ufsMode = ufsStatus.getMode();
     Mode mode = new Mode(ufsMode);
     Long ufsLastModified = ufsStatus.getLastModifiedTime();