HDFS-7179. DFSClient should instantiate a KeyProvider, not a KeyProviderCryptoExtension. (wang)

(cherry picked from commit d2d5a0ea03b0d461a4d376c7b9de8cd5c147effa)
(cherry picked from commit 6ddd9eff78a15081dfe3b405b3d0a958ff27acf5)

Conflicts:
	hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -527,6 +527,9 @@ Release 2.6.0 - UNRELEASED
     HDFS-7178. Additional unit test for replica write with full disk.
     (Arpit Agarwal)
 
+    HDFS-7179. DFSClient should instantiate a KeyProvider, not a
+    KeyProviderCryptoExtension. (wang)
+
     BREAKDOWN OF HDFS-6134 AND HADOOP-10150 SUBTASKS AND RELATED JIRAS
   
       HDFS-6387. HDFS CLI admin tool for creating & deleting an

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -104,6 +104,7 @@ import org.apache.hadoop.crypto.CryptoCodec;
 import org.apache.hadoop.crypto.CryptoInputStream;
 import org.apache.hadoop.crypto.CryptoOutputStream;
 import org.apache.hadoop.crypto.CryptoProtocolVersion;
+import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.BlockStorageLocation;
@@ -264,7 +265,7 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
       new DFSHedgedReadMetrics();
   private static ThreadPoolExecutor HEDGED_READ_THREAD_POOL;
   @VisibleForTesting
-  KeyProviderCryptoExtension provider;
+  KeyProvider provider;
   /**
    * DFSClient configuration 
    */
@@ -596,7 +597,7 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
     this.authority = nameNodeUri == null? "null": nameNodeUri.getAuthority();
     this.clientName = "DFSClient_" + dfsClientConf.taskId + "_" + 
         DFSUtil.getRandom().nextInt()  + "_" + Thread.currentThread().getId();
-    provider = DFSUtil.createKeyProviderCryptoExtension(conf);
+    provider = DFSUtil.createKeyProvider(conf);
     if (LOG.isDebugEnabled()) {
       if (provider == null) {
         LOG.debug("No KeyProvider found.");
@@ -1315,7 +1316,9 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
         feInfo.getKeyName(), feInfo.getEzKeyVersionName(), feInfo.getIV(),
         feInfo.getEncryptedDataEncryptionKey());
     try {
-      return provider.decryptEncryptedKey(ekv);
+      KeyProviderCryptoExtension cryptoProvider = KeyProviderCryptoExtension
+          .createKeyProviderCryptoExtension(provider);
+      return cryptoProvider.decryptEncryptedKey(ekv);
     } catch (GeneralSecurityException e) {
       throw new IOException(e);
     }
@@ -3138,7 +3141,7 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
     return HEDGED_READ_METRIC;
   }
 
-  public KeyProviderCryptoExtension getKeyProvider() {
+  public KeyProvider getKeyProvider() {
     return provider;
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSUtil.java

@@ -1825,15 +1825,14 @@ public class DFSUtil {
   }
 
   /**
-   * Creates a new KeyProviderCryptoExtension by wrapping the
-   * KeyProvider specified in the given Configuration.
+   * Creates a new KeyProvider from the given Configuration.
    *
    * @param conf Configuration
-   * @return new KeyProviderCryptoExtension, or null if no provider was found.
+   * @return new KeyProvider, or null if no provider was found.
    * @throws IOException if the KeyProvider is improperly specified in
    *                             the Configuration
    */
-  public static KeyProviderCryptoExtension createKeyProviderCryptoExtension(
+  public static KeyProvider createKeyProvider(
       final Configuration conf) throws IOException {
     final String providerUriStr =
         conf.get(DFSConfigKeys.DFS_ENCRYPTION_KEY_PROVIDER_URI, null);
@@ -1857,6 +1856,24 @@ public class DFSUtil {
       throw new IOException("KeyProvider " + keyProvider.toString()
           + " was found but it is a transient provider.");
     }
+    return keyProvider;
+  }
+
+  /**
+   * Creates a new KeyProviderCryptoExtension by wrapping the
+   * KeyProvider specified in the given Configuration.
+   *
+   * @param conf Configuration
+   * @return new KeyProviderCryptoExtension, or null if no provider was found.
+   * @throws IOException if the KeyProvider is improperly specified in
+   *                             the Configuration
+   */
+  public static KeyProviderCryptoExtension createKeyProviderCryptoExtension(
+      final Configuration conf) throws IOException {
+    KeyProvider keyProvider = createKeyProvider(conf);
+    if (keyProvider == null) {
+      return null;
+    }
     KeyProviderCryptoExtension cryptoProvider = KeyProviderCryptoExtension
         .createKeyProviderCryptoExtension(keyProvider);
     return cryptoProvider;

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

@@ -43,7 +43,6 @@ import org.apache.hadoop.crypto.CipherSuite;
 import org.apache.hadoop.crypto.CryptoProtocolVersion;
 import org.apache.hadoop.crypto.key.JavaKeyStoreProvider;
 import org.apache.hadoop.crypto.key.KeyProvider;
-import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
 import org.apache.hadoop.crypto.key.KeyProviderFactory;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.CreateFlag;
@@ -1043,7 +1042,7 @@ public class TestEncryptionZones {
   public void testDelegationToken() throws Exception {
     UserGroupInformation.createRemoteUser("JobTracker");
     DistributedFileSystem dfs = cluster.getFileSystem();
-    KeyProviderCryptoExtension keyProvider = Mockito.mock(KeyProviderCryptoExtension.class,
+    KeyProvider keyProvider = Mockito.mock(KeyProvider.class,
         withSettings().extraInterfaces(
             DelegationTokenExtension.class,
             CryptoExtension.class));