diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
index 39b35980636..471e6b9fc77 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
@@ -902,6 +902,36 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     return null;
   }
 
+  /**
+   * Locate DefaultAuditLogger, if any, to enable/disable CallerContext.
+   *
+   * @param value
+   *          true, enable CallerContext, otherwise false to disable it.
+   */
+  void setCallerContextEnabled(final boolean value) {
+    for (AuditLogger logger : auditLoggers) {
+      if (logger instanceof DefaultAuditLogger) {
+        ((DefaultAuditLogger) logger).setCallerContextEnabled(value);
+        break;
+      }
+    }
+  }
+
+  /**
+   * Get the value indicating if CallerContext is enabled.
+   *
+   * @return true, if CallerContext is enabled, otherwise false, if it's
+   *         disabled.
+   */
+  boolean getCallerContextEnabled() {
+    for (AuditLogger logger : auditLoggers) {
+      if (logger instanceof DefaultAuditLogger) {
+        return ((DefaultAuditLogger) logger).getCallerContextEnabled();
+      }
+    }
+    return false;
+  }
+
   private List<AuditLogger> initAuditLoggers(Configuration conf) {
     // Initialize the custom access loggers if configured.
     Collection<String> alClasses =
@@ -6848,13 +6878,33 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
           }
         };
 
-    private boolean isCallerContextEnabled;
+    private volatile boolean isCallerContextEnabled;
     private int callerContextMaxLen;
     private int callerSignatureMaxLen;
 
     private boolean logTokenTrackingId;
     private Set<String> debugCmdSet = new HashSet<String>();
 
+    /**
+     * Enable or disable CallerContext.
+     *
+     * @param value
+     *          true, enable CallerContext, otherwise false to disable it.
+     */
+    void setCallerContextEnabled(final boolean value) {
+      isCallerContextEnabled = value;
+    }
+
+    /**
+     * Get the value indicating if CallerContext is enabled.
+     *
+     * @return true, if CallerContext is enabled, otherwise false, if it's
+     *         disabled.
+     */
+    boolean getCallerContextEnabled() {
+      return isCallerContextEnabled;
+    }
+
     @Override
     public void initialize(Configuration conf) {
       isCallerContextEnabled = conf.getBoolean(
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
index 1301940734d..b39c84226f6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java
@@ -107,6 +107,8 @@ import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_DEFAULT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
 import static org.apache.hadoop.hdfs.client.HdfsClientConfigKeys.DFS_NAMENODE_RPC_PORT_DEFAULT;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_CALLER_CONTEXT_ENABLED_KEY;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_CALLER_CONTEXT_ENABLED_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_AUTO_FAILOVER_ENABLED_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_AUTO_FAILOVER_ENABLED_KEY;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HA_FENCE_METHODS_KEY;
@@ -277,7 +279,8 @@ public class NameNode extends ReconfigurableBase implements
       .unmodifiableList(Arrays
           .asList(DFS_HEARTBEAT_INTERVAL_KEY,
               DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY,
-              FS_PROTECTED_DIRECTORIES));
+              FS_PROTECTED_DIRECTORIES,
+              HADOOP_CALLER_CONTEXT_ENABLED_KEY));
 
   private static final String USAGE = "Usage: hdfs namenode ["
       + StartupOption.BACKUP.getName() + "] | \n\t["
@@ -2008,7 +2011,9 @@ public class NameNode extends ReconfigurableBase implements
             + datanodeManager.getHeartbeatRecheckInterval());
       }
     case FS_PROTECTED_DIRECTORIES:
-      return getNamesystem().getFSDirectory().setProtectedDirectories(newVal);
+      return reconfProtectedDirectories(newVal);
+    case HADOOP_CALLER_CONTEXT_ENABLED_KEY:
+      return reconfCallerContextEnabled(newVal);
     default:
       break;
     }
@@ -2016,6 +2021,21 @@ public class NameNode extends ReconfigurableBase implements
         .get(property));
   }
 
+  private String reconfProtectedDirectories(String newVal) {
+    return getNamesystem().getFSDirectory().setProtectedDirectories(newVal);
+  }
+
+  private String reconfCallerContextEnabled(String newVal) {
+    Boolean callerContextEnabled;
+    if (newVal == null) {
+      callerContextEnabled = HADOOP_CALLER_CONTEXT_ENABLED_DEFAULT;
+    } else {
+      callerContextEnabled = Boolean.parseBoolean(newVal);
+    }
+    namesystem.setCallerContextEnabled(callerContextEnabled);
+    return Boolean.toString(callerContextEnabled);
+  }
+
   @Override  // ReconfigurableBase
   protected Configuration getNewConf() {
     return new HdfsConfiguration();
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestNameNodeReconfigure.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestNameNodeReconfigure.java
index abdb1ead487..4fd7af66521 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestNameNodeReconfigure.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestNameNodeReconfigure.java
@@ -34,6 +34,8 @@ import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_CALLER_CONTEXT_ENABLED_KEY;
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_CALLER_CONTEXT_ENABLED_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_KEY;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_HEARTBEAT_INTERVAL_DEFAULT;
 import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHECK_INTERVAL_KEY;
@@ -50,13 +52,60 @@ public class TestNameNodeReconfigure {
   public void setUp() throws IOException {
     Configuration conf = new HdfsConfiguration();
     cluster = new MiniDFSCluster.Builder(conf).build();
+    cluster.waitActive();
+  }
+
+  @Test
+  public void testReconfigureCallerContextEnabled()
+      throws ReconfigurationException {
+    final NameNode nameNode = cluster.getNameNode();
+    final FSNamesystem nameSystem = nameNode.getNamesystem();
+
+    // try invalid values
+    nameNode.reconfigureProperty(HADOOP_CALLER_CONTEXT_ENABLED_KEY, "text");
+    assertEquals(HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value", false,
+        nameSystem.getCallerContextEnabled());
+    assertEquals(
+        HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value",
+        false,
+        nameNode.getConf().getBoolean(HADOOP_CALLER_CONTEXT_ENABLED_KEY,
+            HADOOP_CALLER_CONTEXT_ENABLED_DEFAULT));
+
+    // enable CallerContext
+    nameNode.reconfigureProperty(HADOOP_CALLER_CONTEXT_ENABLED_KEY, "true");
+    assertEquals(HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value", true,
+        nameSystem.getCallerContextEnabled());
+    assertEquals(
+        HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value",
+        true,
+        nameNode.getConf().getBoolean(HADOOP_CALLER_CONTEXT_ENABLED_KEY,
+            HADOOP_CALLER_CONTEXT_ENABLED_DEFAULT));
+
+    // disable CallerContext
+    nameNode.reconfigureProperty(HADOOP_CALLER_CONTEXT_ENABLED_KEY, "false");
+    assertEquals(HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value", false,
+        nameSystem.getCallerContextEnabled());
+    assertEquals(
+        HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value",
+        false,
+        nameNode.getConf().getBoolean(HADOOP_CALLER_CONTEXT_ENABLED_KEY,
+            HADOOP_CALLER_CONTEXT_ENABLED_DEFAULT));
+
+    // revert to default
+    nameNode.reconfigureProperty(HADOOP_CALLER_CONTEXT_ENABLED_KEY, null);
+
+    // verify default
+    assertEquals(HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value", false,
+        nameSystem.getCallerContextEnabled());
+    assertEquals(HADOOP_CALLER_CONTEXT_ENABLED_KEY + " has wrong value", null,
+        nameNode.getConf().get(HADOOP_CALLER_CONTEXT_ENABLED_KEY));
   }
 
   /**
    * Test that we can modify configuration properties.
    */
   @Test
-  public void testReconfigure() throws ReconfigurationException {
+  public void testReconfigureHearbeatCheck1() throws ReconfigurationException {
     final NameNode nameNode = cluster.getNameNode();
     final DatanodeManager datanodeManager = nameNode.namesystem
         .getBlockManager().getDatanodeManager();