HCFS support chown

other/java/hdfs/src/main/java/seaweed/hdfs/SeaweedFileSystem.java

@@ -218,6 +218,24 @@ public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
         return seaweedFileSystemStore.getFileStatus(path);
     }
 
+    /**
+     * Set owner of a path (i.e. a file or a directory).
+     * The parameters owner and group cannot both be null.
+     *
+     * @param path  The path
+     * @param owner If it is null, the original username remains unchanged.
+     * @param group If it is null, the original groupname remains unchanged.
+     */
+    @Override
+    public void setOwner(Path path, final String owner, final String group)
+        throws IOException {
+        LOG.debug("setOwner path: {}", path);
+        path = qualify(path);
+
+        seaweedFileSystemStore.setOwner(path, owner, group);
+
+    }
+
     Path qualify(Path path) {
         return path.makeQualified(uri, workingDirectory);
     }

other/java/hdfs/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java

@@ -273,4 +273,36 @@ public class SeaweedFileSystemStore {
             bufferSize,
             readAheadQueueDepth);
     }
+
+    public void setOwner(Path path, String owner, String group) {
+
+        LOG.debug("setOwner path:{} owner:{} group:{}", path, owner, group);
+
+        FilerProto.Entry entry = lookupEntry(path);
+        if (entry == null) {
+            LOG.debug("setOwner path:{} entry:{}", path, entry);
+            return;
+        }
+
+        FilerProto.Entry.Builder entryBuilder = entry.toBuilder();
+        FilerProto.FuseAttributes.Builder attributesBuilder = entry.getAttributes().toBuilder();
+
+        if (owner != null) {
+            attributesBuilder.setUserName(owner);
+        }
+        if (group != null) {
+            attributesBuilder.clearGroupName();
+            attributesBuilder.addGroupName(group);
+        }
+
+        entryBuilder.setAttributes(attributesBuilder);
+
+        LOG.debug("setOwner path:{} entry:{}", path, entryBuilder, owner, group);
+
+        filerGrpcClient.getBlockingStub().updateEntry(FilerProto.UpdateEntryRequest.newBuilder()
+            .setDirectory(getParentDirectory(path))
+            .setEntry(entryBuilder)
+            .build());
+
+    }
 }