HDFS-16495: RBF should prepend the client ip rather than append it.

Fixes #4054

Signed-off-by: Owen O'Malley <oomalley@linkedin.com>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/ipc/CallerContext.java

@@ -116,7 +116,7 @@ public final class CallerContext {
 
   /** The caller context builder. */
   public static final class Builder {
-    private static final String KEY_VALUE_SEPARATOR = ":";
+    public static final String KEY_VALUE_SEPARATOR = ":";
     /**
      * The illegal separators include '\t', '\n', '='.
      * User should not set illegal separator.

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterRpcClient.java

@@ -466,7 +466,7 @@ public class RouterRpcClient {
           + router.getRouterId());
     }
 
-    appendClientIpPortToCallerContextIfAbsent();
+    addClientIpToCallerContext();
 
     Object ret = null;
     if (rpcMonitor != null) {
@@ -588,19 +588,32 @@ public class RouterRpcClient {
   /**
    * For tracking which is the actual client address.
    * It adds trace info "clientIp:ip" and "clientPort:port"
-   * to caller context if they are absent.
+   * in the caller context, removing the old values if they were
+   * already present.
    */
-  private void appendClientIpPortToCallerContextIfAbsent() {
+  private void addClientIpToCallerContext() {
     CallerContext ctx = CallerContext.getCurrent();
     String origContext = ctx == null ? null : ctx.getContext();
     byte[] origSignature = ctx == null ? null : ctx.getSignature();
-    CallerContext.setCurrent(
-        new CallerContext.Builder(origContext, contextFieldSeparator)
-            .appendIfAbsent(CLIENT_IP_STR, Server.getRemoteAddress())
-            .appendIfAbsent(CLIENT_PORT_STR,
+    CallerContext.Builder builder =
+        new CallerContext.Builder("", contextFieldSeparator)
+            .append(CLIENT_IP_STR, Server.getRemoteAddress())
+            .append(CLIENT_PORT_STR,
                 Integer.toString(Server.getRemotePort()))
-            .setSignature(origSignature)
-            .build());
+            .setSignature(origSignature);
+    // Append the original caller context
+    if (origContext != null) {
+      for (String part : origContext.split(contextFieldSeparator)) {
+        String[] keyValue =
+            part.split(CallerContext.Builder.KEY_VALUE_SEPARATOR, 2);
+        if (keyValue.length == 2) {
+          builder.appendIfAbsent(keyValue[0], keyValue[1]);
+        } else if (keyValue.length == 1) {
+          builder.append(keyValue[0]);
+        }
+      }
+    }
+    CallerContext.setCurrent(builder.build());
   }
 
   /**

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

@@ -1950,9 +1950,10 @@ public class TestRouterRpc {
     FsPermission permission = new FsPermission("755");
     routerProtocol.mkdirs(dirPath, permission, false);
 
-    // The audit log should contains "callerContext=clientContext,clientIp:"
-    assertTrue(auditlog.getOutput()
-        .contains("callerContext=clientContext,clientIp:"));
+    // The audit log should contains "callerContext=clientIp:...,clientContext"
+    final String logOutput = auditlog.getOutput();
+    assertTrue(logOutput.contains("callerContext=clientIp:"));
+    assertTrue(logOutput.contains(",clientContext"));
     assertTrue(verifyFileExists(routerFS, dirPath));
   }
 
@@ -1997,9 +1998,9 @@ public class TestRouterRpc {
     // Create a directory via the router.
     routerProtocol.getFileInfo(dirPath);
 
-    // The audit log should contains the original clientIp and clientPort
+    // The audit log should not contain the original clientIp and clientPort
     // set by client.
-    assertTrue(auditLog.getOutput().contains("clientIp:1.1.1.1"));
-    assertTrue(auditLog.getOutput().contains("clientPort:1234"));
+    assertFalse(auditLog.getOutput().contains("clientIp:1.1.1.1"));
+    assertFalse(auditLog.getOutput().contains("clientPort:1234"));
   }
 }

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

@@ -464,9 +464,8 @@ public class TestRouterRpcMultiDestination extends TestRouterRpc {
     for (String line : auditLog.getOutput().split("\n")) {
       if (line.contains(auditFlag)) {
         // assert origin caller context exist in audit log
-        assertTrue(line.contains("callerContext=clientContext"));
-        String callerContext = line.substring(
-            line.indexOf("callerContext=clientContext"));
+        String callerContext = line.substring(line.indexOf("callerContext="));
+        assertTrue(callerContext.contains("clientContext"));
         // assert client ip info exist in caller context
         assertTrue(callerContext.contains(clientIpInfo));
         // assert client ip info appears only once in caller context