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

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();