HADOOP-14251. Credential provider should handle property key deprecation. Contributed by John Zhuge.

(cherry picked from commit 7e6463d2fb5f9383d88baec290461868cf476e4c)

Conflicts:
	hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java

(cherry picked from commit 832b0b82e269da0529a0584788ef6d6c803fd156)

Conflicts:
	hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/conf/Configuration.java

@@ -289,18 +289,25 @@ public class Configuration implements Iterable<Map.Entry<String,String>>,
       this.customMessage = customMessage;
     }
 
+    private final String getWarningMessage(String key) {
+      return getWarningMessage(key, null);
+    }
+
     /**
      * Method to provide the warning message. It gives the custom message if
      * non-null, and default message otherwise.
      * @param key the associated deprecated key.
+     * @param source the property source.
      * @return message that is to be logged when a deprecated key is used.
      */
-    private final String getWarningMessage(String key) {
+    private String getWarningMessage(String key, String source) {
       String warningMessage;
       if(customMessage == null) {
         StringBuilder message = new StringBuilder(key);
-        String deprecatedKeySuffix = " is deprecated. Instead, use ";
-        message.append(deprecatedKeySuffix);
+        if (source != null) {
+          message.append(" in " + source);
+        }
+        message.append(" is deprecated. Instead, use ");
         for (int i = 0; i < newKeys.length; i++) {
           message.append(newKeys[i]);
           if(i != newKeys.length-1) {
@@ -576,6 +583,14 @@ public class Configuration implements Iterable<Map.Entry<String,String>>,
     return deprecationContext.get().getDeprecatedKeyMap().containsKey(key);
   }
 
+  private static String getDeprecatedKey(String key) {
+    return deprecationContext.get().getReverseDeprecatedKeyMap().get(key);
+  }
+
+  private static DeprecatedKeyInfo getDeprecatedKeyInfo(String key) {
+    return deprecationContext.get().getDeprecatedKeyMap().get(key);
+  }
+
   /**
    * Sets all deprecated properties that are not currently set but have a
    * corresponding new property that is set. Useful for iterating the
@@ -1199,6 +1214,13 @@ public class Configuration implements Iterable<Map.Entry<String,String>>,
     LOG_DEPRECATION.info(message);
   }
 
+  void logDeprecationOnce(String name, String source) {
+    DeprecatedKeyInfo keyInfo = getDeprecatedKeyInfo(name);
+    if (keyInfo != null && !keyInfo.getAndSetAccessed()) {
+      LOG_DEPRECATION.info(keyInfo.getWarningMessage(name, source));
+    }
+  }
+
   /**
    * Unset a previously set property.
    */
@@ -1970,6 +1992,47 @@ public class Configuration implements Iterable<Map.Entry<String,String>>,
     return pass;
   }
 
+  /**
+   * Get the credential entry by name from a credential provider.
+   *
+   * Handle key deprecation.
+   *
+   * @param provider a credential provider
+   * @param name alias of the credential
+   * @return the credential entry or null if not found
+   */
+  private CredentialEntry getCredentialEntry(CredentialProvider provider,
+                                             String name) throws IOException {
+    CredentialEntry entry = provider.getCredentialEntry(name);
+    if (entry != null) {
+      return entry;
+    }
+
+    // The old name is stored in the credential provider.
+    String oldName = getDeprecatedKey(name);
+    if (oldName != null) {
+      entry = provider.getCredentialEntry(oldName);
+      if (entry != null) {
+        logDeprecationOnce(oldName, provider.toString());
+        return entry;
+      }
+    }
+
+    // The name is deprecated.
+    DeprecatedKeyInfo keyInfo = getDeprecatedKeyInfo(name);
+    if (keyInfo != null && keyInfo.newKeys != null) {
+      for (String newName : keyInfo.newKeys) {
+        entry = provider.getCredentialEntry(newName);
+        if (entry != null) {
+          logDeprecationOnce(name, null);
+          return entry;
+        }
+      }
+    }
+
+    return null;
+  }
+
   /**
    * Try and resolve the provided element name as a credential provider
    * alias.
@@ -1987,7 +2050,7 @@ public class Configuration implements Iterable<Map.Entry<String,String>>,
       if (providers != null) {
         for (CredentialProvider provider : providers) {
           try {
-            CredentialEntry entry = provider.getCredentialEntry(name);
+            CredentialEntry entry = getCredentialEntry(provider, name);
             if (entry != null) {
               pass = entry.getCredential();
               break;

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/conf/TestConfiguration.java

@@ -29,6 +29,7 @@ import java.io.OutputStreamWriter;
 import java.io.StringWriter;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -45,15 +46,22 @@ import static org.junit.Assert.assertArrayEquals;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.hadoop.conf.Configuration.IntegerRanges;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
+import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.net.NetUtils;
+import org.apache.hadoop.security.alias.CredentialProvider;
+import org.apache.hadoop.security.alias.CredentialProviderFactory;
+import org.apache.hadoop.security.alias.LocalJavaKeyStoreProvider;
 import org.apache.hadoop.test.GenericTestUtils;
 
 import static org.apache.hadoop.util.PlatformName.IBM_JAVA;
 import static org.junit.Assert.fail;
 
 import org.codehaus.jackson.map.ObjectMapper;
+import org.hamcrest.CoreMatchers;
+import org.junit.Assert;
 
 public class TestConfiguration extends TestCase {
 
@@ -1403,6 +1411,60 @@ public class TestConfiguration extends TestCase {
             + classes.length, 0, classes.length);
   }
   
+  public void testGetPasswordDeprecatedKeyStored() throws Exception {
+    final String oldKey = "test.password.old.key";
+    final String newKey = "test.password.new.key";
+    final String password = "MyPasswordForDeprecatedKey";
+
+    final File tmpDir = GenericTestUtils.getRandomizedTestDir();
+    tmpDir.mkdirs();
+    final String ourUrl = new URI(LocalJavaKeyStoreProvider.SCHEME_NAME,
+        "file",  new File(tmpDir, "test.jks").toString(), null).toString();
+
+    conf = new Configuration(false);
+    conf.set(CredentialProviderFactory.CREDENTIAL_PROVIDER_PATH, ourUrl);
+    CredentialProvider provider =
+        CredentialProviderFactory.getProviders(conf).get(0);
+    provider.createCredentialEntry(oldKey, password.toCharArray());
+    provider.flush();
+
+    Configuration.addDeprecation(oldKey, newKey);
+
+    Assert.assertThat(conf.getPassword(newKey),
+        CoreMatchers.is(password.toCharArray()));
+    Assert.assertThat(conf.getPassword(oldKey),
+        CoreMatchers.is(password.toCharArray()));
+
+    FileUtil.fullyDelete(tmpDir);
+  }
+
+  public void testGetPasswordByDeprecatedKey() throws Exception {
+    final String oldKey = "test.password.old.key";
+    final String newKey = "test.password.new.key";
+    final String password = "MyPasswordForDeprecatedKey";
+
+    final File tmpDir = GenericTestUtils.getRandomizedTestDir();
+    tmpDir.mkdirs();
+    final String ourUrl = new URI(LocalJavaKeyStoreProvider.SCHEME_NAME,
+        "file",  new File(tmpDir, "test.jks").toString(), null).toString();
+
+    conf = new Configuration(false);
+    conf.set(CredentialProviderFactory.CREDENTIAL_PROVIDER_PATH, ourUrl);
+    CredentialProvider provider =
+        CredentialProviderFactory.getProviders(conf).get(0);
+    provider.createCredentialEntry(newKey, password.toCharArray());
+    provider.flush();
+
+    Configuration.addDeprecation(oldKey, newKey);
+
+    Assert.assertThat(conf.getPassword(newKey),
+        CoreMatchers.is(password.toCharArray()));
+    Assert.assertThat(conf.getPassword(oldKey),
+        CoreMatchers.is(password.toCharArray()));
+
+    FileUtil.fullyDelete(tmpDir);
+  }
+
   public static void main(String[] argv) throws Exception {
     junit.textui.TestRunner.main(new String[]{
       TestConfiguration.class.getName()