HDFS-13941. make storageId in BlockPoolTokenSecretManager.checkAccess optional. Contributed by Ajay Kumar.

(cherry picked from commit c58811c77d5c0442c404a5b2876e09eaf6d16073)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockPoolTokenSecretManager.java

@@ -94,6 +94,18 @@ public class BlockPoolTokenSecretManager extends
         storageTypes, storageIds);
   }
 
+  /**
+   * See {@link BlockTokenSecretManager#checkAccess(BlockTokenIdentifier,
+   * String, ExtendedBlock, BlockTokenIdentifier.AccessMode,
+   * StorageType[])}
+   */
+  public void checkAccess(BlockTokenIdentifier id, String userId,
+      ExtendedBlock block, AccessMode mode, StorageType[] storageTypes)
+      throws InvalidToken {
+    get(block.getBlockPoolId()).checkAccess(id, userId, block, mode,
+        storageTypes);
+  }
+
   /**
    * See {@link BlockTokenSecretManager#checkAccess(Token, String,
    *                ExtendedBlock, BlockTokenIdentifier.AccessMode,
@@ -108,7 +120,7 @@ public class BlockPoolTokenSecretManager extends
   }
 
   /**
-   * See {@link BlockTokenSecretManager#addKeys(ExportedBlockKeys)}
+   * See {@link BlockTokenSecretManager#addKeys(ExportedBlockKeys)}.
    */
   public void addKeys(String bpid, ExportedBlockKeys exportedKeys)
       throws IOException {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java

@@ -295,6 +295,23 @@ public class BlockTokenSecretManager extends
     }
   }
 
+  /**
+   * Check if access should be allowed. userID is not checked if null. This
+   * method doesn't check if token password is correct. It should be used only
+   * when token password has already been verified (e.g., in the RPC layer).
+   *
+   * Some places need to check the access using StorageTypes and for other
+   * places the StorageTypes is not relevant.
+   */
+  public void checkAccess(BlockTokenIdentifier id, String userId,
+      ExtendedBlock block, BlockTokenIdentifier.AccessMode mode,
+      StorageType[] storageTypes) throws InvalidToken {
+    checkAccess(id, userId, block, mode);
+    if (ArrayUtils.isNotEmpty(storageTypes)) {
+      checkAccess(id.getStorageTypes(), storageTypes, "StorageTypes");
+    }
+  }
+
   public void checkAccess(BlockTokenIdentifier id, String userId,
       ExtendedBlock block, BlockTokenIdentifier.AccessMode mode)
       throws InvalidToken {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java

@@ -216,7 +216,11 @@ public class TestBlockToken {
   private static void checkAccess(BlockTokenSecretManager m,
       Token<BlockTokenIdentifier> t, ExtendedBlock blk,
       BlockTokenIdentifier.AccessMode mode, StorageType[] storageTypes,
-      String[] storageIds) throws SecretManager.InvalidToken {
+      String[] storageIds) throws IOException {
+    if(storageIds == null) {
+      // Test overloaded checkAccess method.
+      m.checkAccess(t.decodeIdentifier(), null, blk, mode, storageTypes);
+    }
     m.checkAccess(t, null, blk, mode, storageTypes, storageIds);
   }
 
@@ -802,6 +806,7 @@ public class TestBlockToken {
         emptyStorageIds);
     sm.checkAccess(id, null, block3, mode, storageTypes,
         null);
+    sm.checkAccess(id, null, block3, mode, storageTypes);
   }
 
   @Test