HADOOP-11174. Delegation token for KMS should only be got once if it already exists. (Yi Liu via wang)

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -586,6 +586,9 @@ Release 2.6.0 - UNRELEASED
 
     HADOOP-11178. Fix findbugs exclude file. (Arun Suresh via wang)
 
+    HADOOP-11174. Delegation token for KMS should only be got once if it
+    already exists. (Yi Liu via wang)
+
   OPTIMIZATIONS
 
     HADOOP-10838. Byte array native checksumming. (James Thomas via todd)

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

@@ -26,8 +26,10 @@
 import org.apache.hadoop.crypto.key.KeyProviderFactory;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.ProviderUtils;
+import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
 import org.apache.hadoop.security.authentication.client.AuthenticationException;
@@ -48,6 +50,7 @@
 import java.io.Writer;
 import java.lang.reflect.UndeclaredThrowableException;
 import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
 import java.net.SocketTimeoutException;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -773,23 +776,35 @@ public void drain(String keyName) {
   @Override
   public Token<?>[] addDelegationTokens(String renewer,
       Credentials credentials) throws IOException {
-    Token<?>[] tokens;
-    URL url = createURL(null, null, null, null);
-    DelegationTokenAuthenticatedURL authUrl =
-        new DelegationTokenAuthenticatedURL(configurator);
-    try {
-      Token<?> token = authUrl.getDelegationToken(url, authToken, renewer);
-      if (token != null) {
-        credentials.addToken(token.getService(), token);
-        tokens = new Token<?>[] { token };
-      } else {
-        throw new IOException("Got NULL as delegation token");
+    Token<?>[] tokens = null;
+    Text dtService = getDelegationTokenService();
+    Token<?> token = credentials.getToken(dtService);
+    if (token == null) {
+      URL url = createURL(null, null, null, null);
+      DelegationTokenAuthenticatedURL authUrl =
+          new DelegationTokenAuthenticatedURL(configurator);
+      try {
+        token = authUrl.getDelegationToken(url, authToken, renewer);
+        if (token != null) {
+          credentials.addToken(token.getService(), token);
+          tokens = new Token<?>[] { token };
+        } else {
+          throw new IOException("Got NULL as delegation token");
+        }
+      } catch (AuthenticationException ex) {
+        throw new IOException(ex);
       }
-    } catch (AuthenticationException ex) {
-      throw new IOException(ex);
     }
     return tokens;
   }
+  
+  private Text getDelegationTokenService() throws IOException {
+    URL url = new URL(kmsUrl);
+    InetSocketAddress addr = new InetSocketAddress(url.getHost(),
+        url.getPort());
+    Text dtService = SecurityUtil.buildTokenService(addr);
+    return dtService;
+  }
 
   /**
    * Shutdown valueQueue executor threads

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

@@ -118,7 +118,7 @@ public class TestEncryptionZones {
 
   private MiniDFSCluster cluster;
   private HdfsAdmin dfsAdmin;
-  private DistributedFileSystem fs;
+  protected DistributedFileSystem fs;
   private File testRootDir;
   private final String TEST_KEY = "testKey";
 
@@ -149,12 +149,16 @@ public void setup() throws Exception {
     fcWrapper = new FileContextTestWrapper(
         FileContext.getFileContext(cluster.getURI(), conf));
     dfsAdmin = new HdfsAdmin(cluster.getURI(), conf);
+    setProvider();
+    // Create a test key
+    DFSTestUtil.createKey(TEST_KEY, cluster, conf);
+  }
+  
+  protected void setProvider() {
     // Need to set the client's KeyProvider to the NN's for JKS,
     // else the updates do not get flushed properly
     fs.getClient().provider = cluster.getNameNode().getNamesystem()
         .getProvider();
-    // Create a test key
-    DFSTestUtil.createKey(TEST_KEY, cluster, conf);
   }
 
   @After

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

@@ -19,11 +19,16 @@
 
 import org.apache.hadoop.crypto.key.kms.KMSClientProvider;
 import org.apache.hadoop.crypto.key.kms.server.MiniKMS;
+import org.apache.hadoop.security.Credentials;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
+import org.junit.Test;
 
 import java.io.File;
+import java.util.Arrays;
 import java.util.UUID;
 
 public class TestEncryptionZonesWithKMS extends TestEncryptionZones {
@@ -52,5 +57,26 @@ public void teardown() {
     super.teardown();
     miniKMS.stop();
   }
+  
+  @Override
+  protected void setProvider() {
+  }
 
+  @Test(timeout = 120000)
+  public void testDelegationToken() throws Exception {
+    final String renewer = "JobTracker";
+    UserGroupInformation.createRemoteUser(renewer);
+
+    Credentials creds = new Credentials();
+    Token<?> tokens[] = fs.addDelegationTokens(renewer, creds);
+    DistributedFileSystem.LOG.debug("Delegation tokens: " +
+        Arrays.asList(tokens));
+    Assert.assertEquals(2, tokens.length);
+    Assert.assertEquals(2, creds.numberOfTokens());
+    
+    // If the dt exists, will not get again
+    tokens = fs.addDelegationTokens(renewer, creds);
+    Assert.assertEquals(0, tokens.length);
+    Assert.assertEquals(2, creds.numberOfTokens());
+  }
 }