HDFS-17500: Add missing operation name while authorizing some operations (apache#6776). Contributed by kulkabhay.

Signed-off-by: He Xiaoqiao <[email protected]>

Author: kulkabhay

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

@@ -2618,9 +2618,10 @@ void unsetStoragePolicy(String src) throws IOException {
    * @throws  IOException
    */
   BlockStoragePolicy getStoragePolicy(String src) throws IOException {
+    final String operationName = "getStoragePolicy";
     checkOperation(OperationCategory.READ);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
+    FSPermissionChecker.setOperationType(operationName);
     readLock();
     try {
       checkOperation(OperationCategory.READ);
@@ -2646,9 +2647,10 @@ BlockStoragePolicy[] getStoragePolicies() throws IOException {
   }
 
   long getPreferredBlockSize(String src) throws IOException {
+    final String operationName = "getPreferredBlockSize";
     checkOperation(OperationCategory.READ);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
+    FSPermissionChecker.setOperationType(operationName);
     readLock();
     try {
       checkOperation(OperationCategory.READ);
@@ -2709,6 +2711,8 @@ HdfsFileStatus startFile(String src, PermissionStatus permissions,
       String storagePolicy, boolean logRetryCache) throws IOException {
 
     HdfsFileStatus status;
+    final String operationName = "create";
+    FSPermissionChecker.setOperationType(operationName);
     try {
       status = startFileInt(src, permissions, holder, clientMachine, flag,
           createParent, replication, blockSize, supportedVersions, ecPolicyName,
@@ -2764,7 +2768,6 @@ private HdfsFileStatus startFileInt(String src,
 
     checkOperation(OperationCategory.WRITE);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
@@ -2857,9 +2860,10 @@ private HdfsFileStatus startFileInt(String src,
   boolean recoverLease(String src, String holder, String clientMachine)
       throws IOException {
     boolean skipSync = false;
+    final String operationName = "recoverLease";
     checkOperation(OperationCategory.WRITE);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
+    FSPermissionChecker.setOperationType(operationName);
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
@@ -3105,9 +3109,10 @@ LocatedBlock getAdditionalDatanode(String src, long fileId,
     final byte storagePolicyID;
     final List<DatanodeStorageInfo> chosen;
     final BlockType blockType;
+    final String operationName = "getAdditionalDatanode";
     checkOperation(OperationCategory.WRITE);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
+    FSPermissionChecker.setOperationType(operationName);
     readLock();
     try {
       // Changing this operation category to WRITE instead of making getAdditionalDatanode as a
@@ -3155,10 +3160,11 @@ LocatedBlock getAdditionalDatanode(String src, long fileId,
    */
   void abandonBlock(ExtendedBlock b, long fileId, String src, String holder)
       throws IOException {
+    final String operationName = "abandonBlock";
     NameNode.stateChangeLog.debug("BLOCK* NameSystem.abandonBlock: {} of file {}", b, src);
     checkOperation(OperationCategory.WRITE);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
+    FSPermissionChecker.setOperationType(operationName);
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
@@ -3222,9 +3228,10 @@ boolean completeFile(final String src, String holder,
                        ExtendedBlock last, long fileId)
     throws IOException {
     boolean success = false;
+    final String operationName = "completeFile";
     checkOperation(OperationCategory.WRITE);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
+    FSPermissionChecker.setOperationType(operationName);
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);
@@ -3666,10 +3673,11 @@ void setQuota(String src, long nsQuota, long ssQuota, StorageType type)
    */
   void fsync(String src, long fileId, String clientName, long lastBlockLength)
       throws IOException {
+    final String operationName = "fsync";
     NameNode.stateChangeLog.info("BLOCK* fsync: " + src + " for " + clientName);
     checkOperation(OperationCategory.WRITE);
     final FSPermissionChecker pc = getPermissionChecker();
-    FSPermissionChecker.setOperationType(null);
+    FSPermissionChecker.setOperationType(operationName);
     writeLock();
     try {
       checkOperation(OperationCategory.WRITE);