HDFS-15091. Cache Admin and Quota Commands Should Check SuperUser Before Taking Lock. Contributed by Ayush Saxena.

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

@@ -233,9 +233,6 @@ public class FSDirAttrOp {
    */
   static void setQuota(FSDirectory fsd, FSPermissionChecker pc, String src,
       long nsQuota, long ssQuota, StorageType type) throws IOException {
-    if (fsd.isPermissionEnabled()) {
-      pc.checkSuperuserPrivilege();
-    }
 
     fsd.writeLock();
     try {

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

@@ -78,11 +78,6 @@ class FSNDNCacheOp {
       FSNamesystem fsn, CacheManager cacheManager, CachePoolInfo req,
       boolean logRetryCache)
       throws IOException {
-    final FSPermissionChecker pc = getFsPermissionChecker(fsn);
-
-    if (pc != null) {
-      pc.checkSuperuserPrivilege();
-    }
     CachePoolInfo info = cacheManager.addCachePool(req);
     fsn.getEditLog().logAddCachePool(info, logRetryCache);
     return info;
@@ -91,10 +86,6 @@ class FSNDNCacheOp {
   static void modifyCachePool(
       FSNamesystem fsn, CacheManager cacheManager, CachePoolInfo req,
       boolean logRetryCache) throws IOException {
-    final FSPermissionChecker pc = getFsPermissionChecker(fsn);
-    if (pc != null) {
-      pc.checkSuperuserPrivilege();
-    }
     cacheManager.modifyCachePool(req);
     fsn.getEditLog().logModifyCachePool(req, logRetryCache);
   }
@@ -102,10 +93,6 @@ class FSNDNCacheOp {
   static void removeCachePool(
       FSNamesystem fsn, CacheManager cacheManager, String cachePoolName,
       boolean logRetryCache) throws IOException {
-    final FSPermissionChecker pc = getFsPermissionChecker(fsn);
-    if (pc != null) {
-      pc.checkSuperuserPrivilege();
-    }
     cacheManager.removeCachePool(cachePoolName);
     fsn.getEditLog().logRemoveCachePool(cachePoolName, logRetryCache);
   }

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

@@ -3378,6 +3378,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     final String operationName = getQuotaCommand(nsQuota, ssQuota);
     final FSPermissionChecker pc = getPermissionChecker();
     try {
+      checkSuperuserPrivilege(pc);
       writeLock();
       try {
         checkOperation(OperationCategory.WRITE);
@@ -7197,6 +7198,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     checkOperation(OperationCategory.WRITE);
     String poolInfoStr = null;
     try {
+      checkSuperuserPrivilege();
       writeLock();
       try {
         checkOperation(OperationCategory.WRITE);
@@ -7223,6 +7225,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     String poolNameStr = "{poolName: " +
         (req == null ? null : req.getPoolName()) + "}";
     try {
+      checkSuperuserPrivilege();
       writeLock();
       try {
         checkOperation(OperationCategory.WRITE);
@@ -7248,6 +7251,7 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
     checkOperation(OperationCategory.WRITE);
     String poolNameStr = "{poolName: " + cachePoolName + "}";
     try {
+      checkSuperuserPrivilege();
       writeLock();
       try {
         checkOperation(OperationCategory.WRITE);