HDFS-14026. Overload BlockPoolTokenSecretManager.checkAccess to make storageId and storageType optional. Contributed by Arpit Agarwal.

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

@@ -106,6 +106,26 @@ public class BlockPoolTokenSecretManager extends
         storageTypes);
   }
 
+  /**
+   * See {@link BlockTokenSecretManager#checkAccess(BlockTokenIdentifier,
+   * String, ExtendedBlock, BlockTokenIdentifier.AccessMode)}.
+   */
+  public void checkAccess(BlockTokenIdentifier id, String userId,
+                          ExtendedBlock block, AccessMode mode)
+      throws InvalidToken {
+    get(block.getBlockPoolId()).checkAccess(id, userId, block, mode);
+  }
+
+  /**
+   * See {@link BlockTokenSecretManager#checkAccess(Token, String,
+   *                ExtendedBlock, BlockTokenIdentifier.AccessMode)}.
+   */
+  public void checkAccess(Token<BlockTokenIdentifier> token,
+      String userId, ExtendedBlock block, AccessMode mode)
+      throws InvalidToken {
+    get(block.getBlockPoolId()).checkAccess(token, userId, block, mode);
+  }
+
   /**
    * See {@link BlockTokenSecretManager#checkAccess(Token, String,
    *                ExtendedBlock, BlockTokenIdentifier.AccessMode,

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

@@ -390,6 +390,26 @@ public class BlockTokenSecretManager extends
     }
   }
 
+  /** Check if access should be allowed. userID is not checked if null */
+  public void checkAccess(Token<BlockTokenIdentifier> token, String userId,
+      ExtendedBlock block, BlockTokenIdentifier.AccessMode mode)
+      throws InvalidToken {
+    BlockTokenIdentifier id = new BlockTokenIdentifier();
+    try {
+      id.readFields(new DataInputStream(new ByteArrayInputStream(token
+          .getIdentifier())));
+    } catch (IOException e) {
+      throw new InvalidToken(
+          "Unable to de-serialize block token identifier for user=" + userId
+              + ", block=" + block + ", access mode=" + mode);
+    }
+    checkAccess(id, userId, block, mode);
+    if (!Arrays.equals(retrievePassword(id), token.getPassword())) {
+      throw new InvalidToken("Block token with " + id
+          + " doesn't have the correct token password");
+    }
+  }
+
   private static boolean isExpired(long expiryDate) {
     return Time.now() > expiryDate;
   }

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

@@ -220,6 +220,11 @@ public class TestBlockToken {
     if (storageIds == null) {
       // Test overloaded checkAccess method.
       m.checkAccess(t.decodeIdentifier(), null, blk, mode, storageTypes);
+
+      if (storageTypes == null) {
+        // Test overloaded checkAccess method.
+        m.checkAccess(t, null, blk, mode);
+      }
     }
     m.checkAccess(t, null, blk, mode, storageTypes, storageIds);
   }
@@ -807,6 +812,7 @@ public class TestBlockToken {
     sm.checkAccess(id, null, block3, mode, storageTypes,
         null);
     sm.checkAccess(id, null, block3, mode, storageTypes);
+    sm.checkAccess(id, null, block3, mode);
   }
 
   @Test