From c58811c77d5c0442c404a5b2876e09eaf6d16073 Mon Sep 17 00:00:00 2001
From: Arpit Agarwal <arp@apache.org>
Date: Mon, 22 Oct 2018 14:44:28 -0700
Subject: [PATCH] HDFS-13941. make storageId in
 BlockPoolTokenSecretManager.checkAccess optional. Contributed by Ajay Kumar.

---
 .../block/BlockPoolTokenSecretManager.java      | 14 +++++++++++++-
 .../token/block/BlockTokenSecretManager.java    | 17 +++++++++++++++++
 .../security/token/block/TestBlockToken.java    |  7 ++++++-
 3 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockPoolTokenSecretManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockPoolTokenSecretManager.java
index 8400b4f04b..4d3915e3fa 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockPoolTokenSecretManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockPoolTokenSecretManager.java
@@ -94,6 +94,18 @@ public void checkAccess(BlockTokenIdentifier id, String userId,
         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 void checkAccess(Token<BlockTokenIdentifier> token,
   }
 
   /**
-   * See {@link BlockTokenSecretManager#addKeys(ExportedBlockKeys)}
+   * See {@link BlockTokenSecretManager#addKeys(ExportedBlockKeys)}.
    */
   public void addKeys(String bpid, ExportedBlockKeys exportedKeys)
       throws IOException {
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java
index 21fbbe41d8..85fef13bd3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/BlockTokenSecretManager.java
@@ -295,6 +295,23 @@ public void checkAccess(BlockTokenIdentifier id, String userId,
     }
   }
 
+  /**
+   * 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 {
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
index a8f424005f..4bdd34c2af 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/security/token/block/TestBlockToken.java
@@ -216,7 +216,11 @@ public void testWritableProtobuf() throws Exception {
   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 @@ private void testBadStorageIDCheckAccess(boolean enableProtobuf)
         emptyStorageIds);
     sm.checkAccess(id, null, block3, mode, storageTypes,
         null);
+    sm.checkAccess(id, null, block3, mode, storageTypes);
   }
 
   @Test