feat: repartition mount table with consistency

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/nnproxy/server/mount/MountsManager.java

@@ -16,6 +16,13 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.nnproxy.ProxyConfig;
 import org.apache.hadoop.service.AbstractService;
+import org.apache.hadoop.ipc.RPC;
+import org.apache.hadoop.hdfs.server.namenode.FsImageProto.MountPartition;
+import org.apache.hadoop.hdfs.server.namenode.FSMountRepartitionProtocol;
+
+import java.io.OutputStream;
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -327,6 +334,45 @@ public void load(String mounts) throws Exception {
     }
   }
 
+  public void repartition(String mounts) throws Exception {
+    if (framework.checkExists().forPath(zkMountTablePath) == null) {
+      framework.create().forPath(zkMountTablePath, mounts.getBytes());
+    } else {
+      for (String s : mounts.split("\n")) {
+        if (StringUtils.isEmpty(s)) {
+          continue;
+        }
+        String[] cols = s.split(" ");
+        String newUri = cols[0];
+        String mPoint = cols[1];
+        boolean repartPoint = (cols.length > 2) ? true : false;
+
+        // find the old mount point
+        String oldUri = this.lookupMap.get(mPoint).get(0).fsUri;
+
+        // update the local cache in the old destination (NameNode)
+        try {
+          MountPartition mp = MountPartition.newBuilder()
+            .setMountPoint(mPoint)
+            .setOldUri(oldUri)
+            .setNewUri(newUri).build();
+
+          byte[] data = mp.toByteArray();
+          FSMountRepartitionProtocol proxy = (FSMountRepartitionProtocol) RPC.getProxy(
+            FSMountRepartitionProtocol.class, FSMountRepartitionProtocol.versionID,
+            new InetSocketAddress(oldUri, 10086), new Configuration());
+          proxy.recordMove(data);
+        } catch (Exception e) {
+          e.printStackTrace();
+        }
+
+        if (repartPoint) break;
+      }
+      // update mount table in Zookeeper
+      framework.setData().forPath(zkMountTablePath, mounts.getBytes());
+    }
+  }
+
   protected Map<String, List<MountEntry>> buildLookupMap(List<MountEntry> entries) {
     Map<String, List<MountEntry>> lookupMap = new HashMap<>();
     for (MountEntry entry : entries) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/nnproxy/tools/ReparititionMount.java

@@ -0,0 +1,43 @@
+package org.apache.hadoop.hdfs.nnproxy.tools;
+
+import org.apache.hadoop.hdfs.nnproxy.server.mount.MountsManager;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.util.Tool;
+import org.apache.hadoop.util.ToolRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** */
+public class ReparititionMount implements Tool {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ReparititionMount.class);
+
+    Configuration conf;
+
+    public static void main(String[] args) throws Exception {
+        ReparititionMount main = new ReparititionMount();
+        System.exit(ToolRunner.run(new HdfsConfiguration(), main, args));
+    }
+
+    @Override
+    public int run(String[] args) throws Exception {
+        String mounts = IOUtils.toString(System.in);
+        MountsManager mountsManager = new MountsManager();
+        mountsManager.init(conf);
+        mountsManager.start();
+        mountsManager.repartition(mounts);
+        return 0;
+    }
+
+    @Override
+    public void setConf(Configuration conf) {
+        this.conf = conf;
+    }
+
+    @Override
+    public Configuration getConf() {
+        return conf;
+    }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeAdminManager.java

@@ -744,7 +744,7 @@ private void processBlocksInternal(
         // Update various counts
         lowRedundancyBlocks++;
         if (bc.isUnderConstruction()) {
-          INode ucFile = namesystem.getFSDirectory().getInode(bc.getId());
+          INode ucFile = null;
           if (!(ucFile instanceof  INodeFile) ||
               !ucFile.asFile().isUnderConstruction()) {
             LOG.warn("File {} is not under construction. Skipping add to " +

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/EncryptionZoneManager.java

@@ -360,7 +360,7 @@ boolean isInAnEZ(INodesInPath iip) throws UnresolvedLinkException,
    */
   String getFullPathName(Long nodeId) {
     assert dir.hasReadLock();
-    INode inode = dir.getInode(nodeId);
+    INode inode = null;
     if (inode == null) {
       return null;
     }
@@ -629,7 +629,7 @@ private boolean pathResolvesToId(final long zoneId, final String zonePath)
       throws UnresolvedLinkException, AccessControlException,
       ParentNotDirectoryException {
     assert dir.hasReadLock();
-    INode inode = dir.getInode(zoneId);
+    INode inode = null;
     if (inode == null) {
       return false;
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSMountRepartitionProtocol.java

@@ -0,0 +1,28 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode;
+
+import java.io.IOException;
+import org.apache.hadoop.ipc.VersionedProtocol;
+
+import org.apache.hadoop.hdfs.server.namenode.FsImageProto.MountPartition;
+
+public interface FSMountRepartitionProtocol extends VersionedProtocol {
+    public static final long versionID = 1L;
+    public void recordMove(byte[] data) throws IOException;
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSMountRepartitionProtocolImpl.java

@@ -0,0 +1,60 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdfs.server.namenode;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.apache.hadoop.ipc.ProtocolSignature;
+import org.apache.hadoop.ipc.VersionedProtocol;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.ipc.RPC;
+
+import org.apache.hadoop.hdfs.DFSUtil;
+
+import org.apache.hadoop.hdfs.server.namenode.FsImageProto.MountPartition;
+
+import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
+
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import com.google.common.base.Preconditions;
+import com.google.protobuf.InvalidProtocolBufferException;
+
+public class FSMountRepartitionProtocolImpl implements FSMountRepartitionProtocol {
+    @Override
+    public void recordMove(byte[] data) throws IOException {
+        MountPartition mp = null;
+        try {
+            mp = MountPartition.parseFrom(data);
+        } catch (InvalidProtocolBufferException e) {
+            e.printStackTrace();
+        }
+        INodeKeyedObjects.getMoveCache().put(mp.getMountPoint(), mp.getNewUri());
+    }
+
+    @Override
+    public long getProtocolVersion(String s, long l) throws IOException {
+        return FSMountRepartitionProtocol.versionID;
+    }
+
+    @Override
+    public ProtocolSignature getProtocolSignature(String s, long l, int i) throws IOException {
+        return new ProtocolSignature(FSMountRepartitionProtocol.versionID, null);
+    }
+}

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSTreeTraverser.java

@@ -233,7 +233,7 @@ private INode resolvePaths(final long startId, List<byte[]> startAfters)
     // If the readlock was reacquired, we need to resolve the paths again
     // in case things have changed. If our cursor file/dir is changed,
     // continue from the next one.
-    INode zoneNode = dir.getInode(startId);
+    INode zoneNode = dir.getInode(null, null);
     if (zoneNode == null) {
       throw new FileNotFoundException("Zone " + startId + " is deleted.");
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeDirectoryAttributes.java

@@ -56,6 +56,11 @@ public boolean isDirectory() {
       return true;
     }
 
+    @Override
+    public String getPath() {
+      return null;
+    }
+
     @Override
     public boolean metadataEquals(INodeDirectoryAttributes other) {
       return other != null

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeFileAttributes.java

@@ -81,6 +81,11 @@ public short getFileReplication() {
       return HeaderFormat.getReplication(header);
     }
 
+    @Override
+    public String getPath() {
+      return null;
+    }
+
     @Override
     public boolean isStriped() {
       return HeaderFormat.isStriped(header);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeKeyedObjects.java

@@ -3,6 +3,8 @@
 import static java.util.concurrent.TimeUnit.*;
 
 import com.github.benmanes.caffeine.cache.Caffeine;
+import com.github.benmanes.caffeine.cache.*;
+import com.github.benmanes.caffeine.cache.stats.CacheStats;
 import com.github.benmanes.caffeine.cache.RemovalCause;
 import java.util.ArrayList;
 import java.util.Iterator;
@@ -16,6 +18,7 @@
 
 public class INodeKeyedObjects {
   private static IndexedCache<String, INode> cache;
+  private static Cache<String, String> move;
 
   private static Set<String> concurrentUpdateSet;
   private static Set<String> concurrentRenameSet;
@@ -375,4 +378,13 @@ public static IndexedCache<String, INode> getCache() {
     }
     return cache;
   }
+
+  public static Cache<String, String> getMoveCache() {
+    if (move == null) {
+      move =  Caffeine.newBuilder()
+              .expireAfterWrite(1000, TimeUnit.MILLISECONDS)
+              .build();      
+    }
+    return move;
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeSymlink.java

@@ -34,7 +34,7 @@ public class INodeSymlink extends INodeWithAdditionalFields {
 
   INodeSymlink(long id, byte[] name, PermissionStatus permissions,
       long mtime, long atime, String symlink) {
-    super(id, name, permissions, mtime, atime, 0L);
+    super(id, name, permissions, mtime, atime, 0L, null);
     this.symlink = DFSUtil.string2Bytes(symlink);
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java

@@ -261,7 +261,7 @@ public class NameNodeRpcServer implements NamenodeProtocols {
   protected final InetSocketAddress clientRpcAddress;
 
   /** The RPC server that listens to logging requests from other namenodes */
-  protected final RPC.Server editLogRpcServer;
+  protected final RPC.Server mountRepartitionRpcServer;
 
   private final String minimumDataNodeVersion;
 
@@ -549,9 +549,9 @@ public NameNodeRpcServer(Configuration conf, NameNode nn)
       }
     }
 
-    // FSEditLog RPC Server
-    editLogRpcServer = new RPC.Builder(conf).setProtocol(FSEditLogProtocol.class)
-      .setInstance(new FSEditLogProtocolImpl())
+    // FSMountRepartition RPC Server
+    mountRepartitionRpcServer = new RPC.Builder(conf).setProtocol(FSMountRepartitionProtocol.class)
+      .setInstance(new FSMountRepartitionProtocolImpl())
       .setBindAddress("0.0.0.0")
       .setPort(10086)
       .setNumHandlers(handlerCount)
@@ -583,7 +583,7 @@ RPC.Server getServiceRpcServer() {
    */
   void start() {
     clientRpcServer.start();
-    editLogRpcServer.start();
+    mountRepartitionRpcServer.start();
     if (serviceRpcServer != null) {
       serviceRpcServer.start();      
     }
@@ -597,6 +597,7 @@ void start() {
    */
   void join() throws InterruptedException {
     clientRpcServer.join();
+    mountRepartitionRpcServer.join();
     if (serviceRpcServer != null) {
       serviceRpcServer.join();      
     }
@@ -612,6 +613,9 @@ void stop() {
     if (clientRpcServer != null) {
       clientRpcServer.stop();
     }
+    if (mountRepartitionRpcServer != null) {
+      mountRepartitionRpcServer.stop();
+    }
     if (serviceRpcServer != null) {
       serviceRpcServer.stop();
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/ReencryptionHandler.java

@@ -390,7 +390,7 @@ void reencryptEncryptionZone(final long zoneId)
 
     traverser.readLock();
     try {
-      zoneNode = dir.getInode(zoneId);
+      zoneNode = null;
       // start re-encrypting the zone from the beginning
       if (zoneNode == null) {
         LOG.info("Directory with id {} removed during re-encrypt, skipping",