HDFS-17954. For all client requests through Router, NameNode retry cache should Router's clientId & callId key.

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterRetryCacheWithoutProxy.java

@@ -0,0 +1,177 @@
+/**
+ * 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.federation.router;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.ha.HAServiceProtocol;
+import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster;
+import org.apache.hadoop.io.retry.RetryInvocationHandler;
+import org.apache.hadoop.ipc.Client;
+import org.apache.hadoop.security.UserGroupInformation;
+
+import static org.apache.hadoop.hdfs.server.federation.FederationTestUtils.NAMENODES;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestRouterRetryCacheWithoutProxy {
+
+  /** Federated HDFS cluster. */
+  private MiniRouterDFSCluster cluster;
+
+  @BeforeEach
+  public  void setup() throws Exception {
+    Configuration conf = new Configuration();
+    cluster = new MiniRouterDFSCluster(true, 1, conf);
+    cluster.addNamenodeOverrides(conf);
+
+    // Start NNs and DNs and wait until ready
+    cluster.startCluster();
+
+    // Start routers with only an RPC service
+    cluster.startRouters();
+
+    // Register and verify all NNs with all routers
+    cluster.registerNamenodes();
+    cluster.waitNamenodeRegistration();
+
+    // Setup the mount table
+    cluster.installMockLocations();
+
+    // Making one Namenodes active per nameservice
+    if (cluster.isHighAvailability()) {
+      for (String ns : cluster.getNameservices()) {
+        cluster.switchToActive(ns, NAMENODES[0]);
+        cluster.switchToStandby(ns, NAMENODES[1]);
+      }
+    }
+    cluster.waitActiveNamespaces();
+  }
+
+  @AfterEach
+  public void teardown() throws IOException {
+    if (cluster != null) {
+      cluster.shutdown();
+      cluster = null;
+    }
+  }
+
+  @Test
+  public void testRetryCacheWithRouter() throws Exception {
+    RetryInvocationHandler.SET_CALL_ID_FOR_TEST.set(false);
+    FileSystem routerFS = cluster.getRandomRouter().getFileSystem();
+    Path testDir = new Path("/target-ns0/testdir");
+    routerFS.mkdirs(testDir);
+    routerFS.setPermission(testDir, FsPermission.getDefault());
+
+    // Run as fake joe to authorize the test
+    UserGroupInformation joe = UserGroupInformation.createUserForTesting("fake_joe",
+        new String[] {"fake_group"});
+    FileSystem joeFS = joe.doAs((PrivilegedExceptionAction<FileSystem>) () ->
+        FileSystem.newInstance(routerFS.getUri(), routerFS.getConf()));
+
+    Path renameSrc = new Path(testDir, "renameSrc");
+    Path renameDst = new Path(testDir, "renameDst");
+    joeFS.mkdirs(renameSrc);
+
+    assertEquals(HAServiceProtocol.HAServiceState.ACTIVE,
+        cluster.getCluster().getNamesystem(0).getState());
+
+    int callId = Client.nextCallId();
+    Client.setCallIdAndRetryCount(callId, 0, null);
+    assertTrue(joeFS.rename(renameSrc, renameDst));
+
+    Client.setCallIdAndRetryCount(callId, 0, null);
+    assertTrue(joeFS.rename(renameSrc, renameDst));
+
+    String ns0 = cluster.getNameservices().get(0);
+    cluster.switchToStandby(ns0, NAMENODES[0]);
+    cluster.switchToActive(ns0, NAMENODES[1]);
+
+    assertEquals(HAServiceProtocol.HAServiceState.ACTIVE,
+        cluster.getCluster().getNamesystem(1).getState());
+
+    Client.setCallIdAndRetryCount(callId, 0, null);
+    assertTrue(joeFS.rename(renameSrc, renameDst));
+
+    FileStatus fileStatus = joeFS.getFileStatus(renameDst);
+    assertEquals("fake_joe", fileStatus.getOwner());
+
+    joeFS.delete(renameDst, true);
+  }
+
+  @Test
+  public void testRetryCacheWithNameNode() throws Exception {
+    RetryInvocationHandler.SET_CALL_ID_FOR_TEST.set(false);
+    FileSystem fileSystem = cluster.getCluster().getFileSystem(0);
+
+    Path testDir = new Path("/target-ns0/testdir");
+    fileSystem.mkdirs(testDir);
+    fileSystem.setPermission(testDir, FsPermission.getDefault());
+
+    UserGroupInformation joe = UserGroupInformation.createUserForTesting("fake_joe",
+        new String[] {"fake_group"});
+    FileSystem finalRouterFS0 = fileSystem;
+    FileSystem joeFS = joe.doAs((PrivilegedExceptionAction<FileSystem>) () ->
+        FileSystem.newInstance(finalRouterFS0.getUri(), finalRouterFS0.getConf()));
+
+    Path renameSrc = new Path(testDir, "renameSrc");
+    Path renameDst = new Path(testDir, "renameDst");
+    joeFS.mkdirs(renameSrc);
+
+    assertEquals(HAServiceProtocol.HAServiceState.ACTIVE,
+        cluster.getCluster().getNamesystem(0).getState());
+
+    int callId = Client.nextCallId();
+    Client.setCallIdAndRetryCount(callId, 0, null);
+    assertTrue(joeFS.rename(renameSrc, renameDst));
+
+    Client.setCallIdAndRetryCount(callId, 0, null);
+    assertTrue(joeFS.rename(renameSrc, renameDst));
+
+    String ns0 = cluster.getNameservices().get(0);
+    cluster.switchToStandby(ns0, NAMENODES[0]);
+    cluster.switchToActive(ns0, NAMENODES[1]);
+
+    assertEquals(HAServiceProtocol.HAServiceState.ACTIVE,
+        cluster.getCluster().getNamesystem(1).getState());
+
+    fileSystem = cluster.getCluster().getFileSystem(1);
+    FileSystem finalRouterFS1 = fileSystem;
+    joeFS = joe.doAs((PrivilegedExceptionAction<FileSystem>) () ->
+        FileSystem.newInstance(finalRouterFS1.getUri(), finalRouterFS1.getConf()));
+
+    Client.setCallIdAndRetryCount(callId, 0, null);
+    assertTrue(joeFS.rename(renameSrc, renameDst));
+
+    FileStatus fileStatus = joeFS.getFileStatus(renameDst);
+    assertEquals("fake_joe", fileStatus.getOwner());
+
+    joeFS.delete(renameDst, true);
+  }
+}

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

@@ -17,7 +17,6 @@
  */
 package org.apache.hadoop.hdfs.server.namenode;
 
-import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_IP_PROXY_USERS;
 import static org.apache.hadoop.util.ExitUtil.terminate;
 import static org.apache.hadoop.util.Time.monotonicNow;
 
@@ -196,9 +195,6 @@ private enum State {
 
   protected final OpInstanceCache cache = new OpInstanceCache();
 
-  // Users who can override the client ip
-  private final String[] ipProxyUsers;
-
   /**
    * The edit directories that are shared between primary and secondary.
    */
@@ -250,7 +246,6 @@ static FSEditLog newInstance(Configuration conf, NNStorage storage,
    * @param editsDirs List of journals to use
    */
   FSEditLog(Configuration conf, NNStorage storage, List<URI> editsDirs) {
-    ipProxyUsers = conf.getStrings(DFS_NAMENODE_IP_PROXY_USERS);
     isSyncRunning = false;
     this.conf = conf;
     this.storage = storage;
@@ -804,8 +799,7 @@ private void printStatistics(boolean force) {
   /** Record the RPC IDs if necessary */
   private void logRpcIds(FSEditLogOp op, boolean toLogRpcIds) {
     if (toLogRpcIds) {
-      Pair<byte[], Integer> clientIdAndCallId =
-          NameNode.getClientIdAndCallId(this.ipProxyUsers);
+      Pair<byte[], Integer> clientIdAndCallId = NameNode.getClientIdAndCallId();
       op.setRpcClientId(clientIdAndCallId.getLeft());
       op.setRpcCallId(clientIdAndCallId.getRight());
     }

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

@@ -535,20 +535,11 @@ public static NameNodeMetrics getNameNodeMetrics() {
 
   /**
    * Try to obtain the actual client info according to the current user.
-   * @param ipProxyUsers Users who can override client infos
    */
-  private static String clientInfoFromContext(
-      final String[] ipProxyUsers) {
-    if (ipProxyUsers != null) {
-      UserGroupInformation user =
-          UserGroupInformation.getRealUserOrSelf(Server.getRemoteUser());
-      if (user != null &&
-          ArrayUtils.contains(ipProxyUsers, user.getShortUserName())) {
-        CallerContext context = CallerContext.getCurrent();
-        if (context != null && context.isContextValid()) {
-          return context.getContext();
-        }
-      }
+  private static String clientInfoFromContext() {
+    CallerContext context = CallerContext.getCurrent();
+    if (context != null && context.isContextValid()) {
+      return context.getContext();
     }
     return null;
   }
@@ -573,12 +564,12 @@ public static String parseSpecialValue(String content, String key) {
 
   /**
    * Try to obtain the actual client's machine according to the current user.
-   * @param ipProxyUsers Users who can override client infos.
+   *
    * @return The actual client's machine.
    */
-  public static String getClientMachine(final String[] ipProxyUsers) {
+  public static String getClientMachine() {
     String clientMachine = null;
-    String cc = clientInfoFromContext(ipProxyUsers);
+    String cc = clientInfoFromContext();
     if (cc != null) {
       // if the rpc has a caller context of "clientIp:1.2.3.4,CLI",
       // return "1.2.3.4" as the client machine.
@@ -597,16 +588,14 @@ public static String getClientMachine(final String[] ipProxyUsers) {
   }
 
   /**
-   * Try to obtain the actual client's id and call id
-   * according to the current user.
-   * @param ipProxyUsers Users who can override client infos
+   * Try to obtain the actual client's id and call id.
+   *
    * @return The actual client's id and call id.
    */
-  public static Pair<byte[], Integer> getClientIdAndCallId(
-      final String[] ipProxyUsers) {
+  public static Pair<byte[], Integer> getClientIdAndCallId() {
     byte[] clientId = Server.getClientId();
     int callId = Server.getCallId();
-    String cc = clientInfoFromContext(ipProxyUsers);
+    String cc = clientInfoFromContext();
     if (cc != null) {
       String clientIdKey = CallerContext.CLIENT_ID_STR +
           CallerContext.Builder.KEY_VALUE_SEPARATOR;

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

@@ -730,8 +730,7 @@ public NamenodeCommand startCheckpoint(NamenodeRegistration registration)
    * Return the current CacheEntry.
    */
   private CacheEntry getCacheEntry() {
-    Pair<byte[], Integer> clientInfo =
-        NameNode.getClientIdAndCallId(this.ipProxyUsers);
+    Pair<byte[], Integer> clientInfo = NameNode.getClientIdAndCallId();
     return RetryCache.waitForCompletion(
         retryCache, clientInfo.getLeft(), clientInfo.getRight());
   }
@@ -740,8 +739,7 @@ private CacheEntry getCacheEntry() {
    * Return the current CacheEntryWithPayload.
    */
   private CacheEntryWithPayload getCacheEntryWithPayload(Object payload) {
-    Pair<byte[], Integer> clientInfo =
-        NameNode.getClientIdAndCallId(this.ipProxyUsers);
+    Pair<byte[], Integer> clientInfo = NameNode.getClientIdAndCallId();
     return RetryCache.waitForCompletion(retryCache, payload,
         clientInfo.getLeft(), clientInfo.getRight());
   }
@@ -1933,7 +1931,7 @@ private void verifySoftwareVersion(DatanodeRegistration dnReg)
    * Get the actual client's machine.
    */
   private String getClientMachine() {
-    return NameNode.getClientMachine(this.ipProxyUsers);
+    return NameNode.getClientMachine();
   }
 
   @Override