HADOOP-14445. Addendum: Use DelegationTokenIssuer to create KMS delegation tokens that can authenticate to all KMS instances.

(cherry picked from commit b6fc72a025)

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderDelegationTokenExtension.java

@@ -22,7 +22,7 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 
 import java.io.IOException;
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderTokenIssuer.java

@@ -22,7 +22,7 @@ import java.net.URI;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.security.token.org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 
 /**
  * File systems that support Encryption Zones have to implement this interface.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java

@@ -64,7 +64,7 @@ import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 import org.apache.hadoop.util.ClassUtil;
 import org.apache.hadoop.util.DataChecksum;
 import org.apache.hadoop.util.Progressable;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/org/apache/hadoop/security/token/DelegationTokenIssuer.java

@@ -15,13 +15,12 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.hadoop.security.token.org.apache.hadoop.security.token;
+package org.apache.hadoop.security.token;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.Credentials;
-import org.apache.hadoop.security.token.Token;
 
 import java.io.IOException;
 import java.util.ArrayList;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestFilterFileSystem.java

@@ -36,7 +36,7 @@ import org.apache.hadoop.fs.Options.CreateOpts;
 import org.apache.hadoop.fs.Options.Rename;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 import org.apache.hadoop.util.Progressable;
 import org.junit.BeforeClass;
 import org.junit.Test;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java

@@ -25,7 +25,7 @@ import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 import org.apache.hadoop.util.Progressable;
 import org.junit.Assert;
 import org.junit.Test;

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -103,7 +103,7 @@ import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifie
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 import org.apache.hadoop.util.ChunkedArrayList;
 import org.apache.hadoop.util.Progressable;
 

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/web/WebHdfsFileSystem.java

@@ -118,7 +118,7 @@ import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.TokenIdentifier;
 import org.apache.hadoop.security.token.TokenSelector;
 import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSelector;
-import org.apache.hadoop.security.token.org.apache.hadoop.security.token.DelegationTokenIssuer;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 import org.apache.hadoop.util.JsonSerialization;
 import org.apache.hadoop.util.KMSUtil;
 import org.apache.hadoop.util.Progressable;
@@ -173,6 +173,7 @@ public class WebHdfsFileSystem extends FileSystem
   private Set<String> restCsrfMethodsToIgnore;
 
   private DFSOpsCountStatistics storageStatistics;
+  private KeyProvider testProvider;
 
   /**
    * Return the protocol scheme for the FileSystem.
@@ -1950,6 +1951,9 @@ public class WebHdfsFileSystem extends FileSystem
 
   @Override
   public KeyProvider getKeyProvider() throws IOException {
+    if (testProvider != null) {
+      return testProvider;
+    }
     URI keyProviderUri = getKeyProviderUri();
     if (keyProviderUri == null) {
       return null;
@@ -1957,6 +1961,11 @@ public class WebHdfsFileSystem extends FileSystem
     return KMSUtil.createKeyProviderFromUri(getConf(), keyProviderUri);
   }
 
+  @VisibleForTesting
+  public void setTestProvider(KeyProvider kp) {
+    testProvider = kp;
+  }
+
   /**
    * This class is used for opening, reading, and seeking files while using the
    * WebHdfsFileSystem. This class will invoke the retry policy when performing

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestEncryptionZones.java

@@ -97,6 +97,7 @@ import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.DelegationTokenIssuer;
 import org.apache.hadoop.util.DataChecksum;
 import org.apache.hadoop.util.ToolRunner;
 import org.apache.hadoop.crypto.key.KeyProviderDelegationTokenExtension.DelegationTokenExtension;
@@ -118,7 +119,6 @@ import static org.mockito.Matchers.anyLong;
 import static org.mockito.Matchers.anyObject;
 import static org.mockito.Matchers.anyShort;
 import static org.mockito.Mockito.withSettings;
-import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyString;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_DEFAULT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
@@ -1339,11 +1339,13 @@ public class TestEncryptionZones {
     byte[] testIdentifier = "Test identifier for delegation token".getBytes();
 
     @SuppressWarnings("rawtypes")
-    Token<?> testToken = new Token(testIdentifier, new byte[0],
+    Token testToken = new Token(testIdentifier, new byte[0],
         new Text(), new Text());
-    Mockito.when(((DelegationTokenExtension)keyProvider).
+    Mockito.when(((DelegationTokenIssuer)keyProvider).
-        addDelegationTokens(anyString(), (Credentials)any())).
+        getCanonicalServiceName()).thenReturn("service");
-        thenReturn(new Token<?>[] { testToken });
+    Mockito.when(((DelegationTokenIssuer)keyProvider).
+        getDelegationToken(anyString())).
+        thenReturn(testToken);
 
     dfs.getClient().setKeyProvider(keyProvider);
 
@@ -1353,7 +1355,7 @@ public class TestEncryptionZones {
         Arrays.asList(tokens));
     Assert.assertEquals(2, tokens.length);
     Assert.assertEquals(tokens[1], testToken);
-    Assert.assertEquals(1, creds.numberOfTokens());
+    Assert.assertEquals(2, creds.numberOfTokens());
   }
 
   /**
@@ -2106,22 +2108,22 @@ public class TestEncryptionZones {
     Mockito.when(keyProvider.getConf()).thenReturn(conf);
     byte[] testIdentifier = "Test identifier for delegation token".getBytes();
 
-    Token<?> testToken = new Token(testIdentifier, new byte[0],
+    Token testToken = new Token(testIdentifier, new byte[0],
         new Text("kms-dt"), new Text());
-    Mockito.when(((DelegationTokenExtension) keyProvider)
+    Mockito.when(((DelegationTokenIssuer)keyProvider).
-        .addDelegationTokens(anyString(), (Credentials) any()))
+        getCanonicalServiceName()).thenReturn("service");
-        .thenReturn(new Token<?>[] {testToken});
+    Mockito.when(((DelegationTokenIssuer)keyProvider).
-
+        getDelegationToken(anyString())).
-    WebHdfsFileSystem webfsSpy = Mockito.spy(webfs);
+        thenReturn(testToken);
-    Mockito.doReturn(keyProvider).when(webfsSpy).getKeyProvider();
 
+    webfs.setTestProvider(keyProvider);
     Credentials creds = new Credentials();
     final Token<?>[] tokens =
-        webfsSpy.addDelegationTokens("JobTracker", creds);
+        webfs.addDelegationTokens("JobTracker", creds);
 
     Assert.assertEquals(2, tokens.length);
     Assert.assertEquals(tokens[1], testToken);
-    Assert.assertEquals(1, creds.numberOfTokens());
+    Assert.assertEquals(2, creds.numberOfTokens());
   }
 
   /**