HADOOP-16350. Ability to tell HDFS client not to request KMS Information from NameNode. Ccontributed by Greg Senia, Ajay Kumar.

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

@@ -403,4 +403,17 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
   public static final Class<? extends DomainNameResolver>
       HADOOP_DOMAINNAME_RESOLVER_IMPL_DEFAULT =
       DNSDomainNameResolver.class;
+  /*
+   *  Ignore KMS default URI returned from NameNode.
+   *  When set to true, kms uri is searched in the following order:
+   *  1. If there is a mapping in Credential's secrets map for namenode uri.
+   *  2. Fallback to local conf.
+   *  If client choose to ignore KMS uri provided by NameNode then client
+   *  should set KMS URI using 'hadoop.security.key.provider.path' to access
+   *  the right KMS for encrypted files.
+   * */
+  public static final String DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI =
+      "dfs.client.ignore.namenode.default.kms.uri";
+  public static final boolean
+      DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI_DEFAULT = false;
 }

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -3480,4 +3480,18 @@
       with the input domain name of the services by querying the underlying DNS.
     </description>
   </property>
+
+  <property>
+    <name>dfs.client.ignore.namenode.default.kms.uri</name>
+    <value>false</value>
+    <description>
+      Ignore KMS default URI returned from NameNode.
+      When set to true, kms uri is searched in the following order:
+      1. If there is a mapping in Credential's secrets map for namenode uri.
+      2. Fallback to local conf. (i.e hadoop.security.key.provider.path)
+      If client choose to ignore KMS uri provided by NameNode then client
+      should set KMS URI using 'hadoop.security.key.provider.path' to access
+      the right KMS for encrypted files.
+    </description>
+  </property>
 </configuration>

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

@@ -17,6 +17,8 @@
  */
 package org.apache.hadoop.hdfs;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeys.DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI;
+import static org.apache.hadoop.fs.CommonConfigurationKeys.DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI_DEFAULT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_CRYPTO_CODEC_CLASSES_KEY_PREFIX;
 
 import java.io.IOException;
@@ -141,12 +143,20 @@ public static URI getKeyProviderUri(UserGroupInformation ugi,
           URI.create(DFSUtilClient.bytes2String(keyProviderUriBytes));
     }
     if (keyProviderUri == null) {
-      // NN is old and doesn't report provider, so use conf.
+      // Check if NN provided uri is not null and ignore property is false.
-      if (keyProviderUriStr == null) {
+      if (keyProviderUriStr != null && !conf.getBoolean(
-        keyProviderUri = KMSUtil.getKeyProviderUri(conf, keyProviderUriKeyName);
+          DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI,
-      } else if (!keyProviderUriStr.isEmpty()) {
+          DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI_DEFAULT)) {
+        if (!keyProviderUriStr.isEmpty()) {
           keyProviderUri = URI.create(keyProviderUriStr);
         }
+      }
+      // Fallback to configuration.
+      if (keyProviderUri == null) {
+        // Either NN is old and doesn't report provider or ignore NN KMS
+        // provider property is set to true, so use conf.
+        keyProviderUri = KMSUtil.getKeyProviderUri(conf, keyProviderUriKeyName);
+      }
       if (keyProviderUri != null) {
         credentials.addSecretKey(
             credsKey, DFSUtilClient.string2Bytes(keyProviderUri.toString()));

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

@@ -111,6 +111,7 @@
 import org.junit.rules.Timeout;
 import org.mockito.Mockito;
 
+import static org.apache.hadoop.fs.CommonConfigurationKeys.DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI;
 import static org.junit.Assert.assertNotNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyBoolean;
@@ -1950,6 +1951,16 @@ public void testDifferentKMSProviderOnUpgradedNamenode() throws Exception {
     Assert.assertEquals("Key Provider for client and namenode are different",
         namenodeKeyProviderUri, cluster.getFileSystem().getClient()
         .getKeyProviderUri());
+
+    // Ignore the key provider from NN.
+    clusterConf.setBoolean(
+        DFS_CLIENT_IGNORE_NAMENODE_DEFAULT_KMS_URI, true);
+    Assert.assertEquals("Expecting Key Provider for client config",
+        "dummy://foo:bar@test_provider1", cluster.getFileSystem().getClient()
+            .getKeyProviderUri().toString());
+    Assert.assertNotEquals("Key Provider for client and namenode is different",
+        namenodeKeyProviderUri, cluster.getFileSystem().getClient()
+            .getKeyProviderUri().toString());
   }
 
   /**