HADOOP-11071. KMSClientProvider should drain the local generated EEK cache on key rollover. (tucu)

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

@@ -771,6 +771,9 @@ Release 2.6.0 - UNRELEASED
     HADOOP-11073. Credential Provider related Unit Tests Failure on Windows.
     (Xiaoyu Yao via cnauroth)
 
+    HADOOP-11071. KMSClientProvider should drain the local generated EEK cache
+    on key rollover. (tucu)
+
 Release 2.5.1 - UNRELEASED
 
   INCOMPATIBLE CHANGES

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

@@ -178,6 +178,13 @@ public class KeyProviderCryptoExtension extends
     public void warmUpEncryptedKeys(String... keyNames)
         throws IOException;
 
+    /**
+     * Drains the Queue for the provided key.
+     *
+     * @param keyName the key to drain the Queue for
+     */
+    public void drain(String keyName);
+
     /**
      * Generates a key material and encrypts it using the given key version name
      * and initialization vector. The generated key material is of the same
@@ -313,6 +320,10 @@ public class KeyProviderCryptoExtension extends
       // NO-OP since the default version does not cache any keys
     }
 
+    @Override
+    public void drain(String keyName) {
+      // NO-OP since the default version does not cache any keys
+    }
   }
 
   /**

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

@@ -590,7 +590,9 @@ public class KMSClientProvider extends KeyProvider implements CryptoExtension,
     conn.setRequestProperty(CONTENT_TYPE, APPLICATION_JSON_MIME);
     Map response = call(conn, jsonMaterial,
         HttpURLConnection.HTTP_OK, Map.class);
-    return parseJSONKeyVersion(response);
+    KeyVersion keyVersion = parseJSONKeyVersion(response);
+    encKeyVersionQueue.drain(name);
+    return keyVersion;
   }
 
 
@@ -712,6 +714,11 @@ public class KMSClientProvider extends KeyProvider implements CryptoExtension,
     }
   }
 
+  @Override
+  public void drain(String keyName) {
+    encKeyVersionQueue.drain(keyName);
+  }
+
   @Override
   public Token<?>[] addDelegationTokens(String renewer,
       Credentials credentials) throws IOException {

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

@@ -227,6 +227,19 @@ public class ValueQueue <E> {
     return getAtMost(keyName, 1).get(0);
   }
 
+  /**
+   * Drains the Queue for the provided key.
+   *
+   * @param keyName the key to drain the Queue for
+   */
+  public void drain(String keyName ) {
+    try {
+      keyQueues.get(keyName).clear();
+    } catch (ExecutionException ex) {
+      //NOP
+    }
+  }
+
   /**
    * This removes the "num" values currently at the head of the Queue for the
    * provided key. Will immediately fire the Queue filler function if key

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/key/TestValueQueue.java

@@ -187,4 +187,18 @@ public class TestValueQueue {
     Assert.assertEquals(10, filler.getTop().num);
     vq.shutdown();
   }
+
+  @Test
+  public void testDrain() throws Exception {
+    MockFiller filler = new MockFiller();
+    ValueQueue<String> vq =
+        new ValueQueue<String>(10, 0.1f, 300, 1,
+            SyncGenerationPolicy.ALL, filler);
+    Assert.assertEquals("test", vq.getNext("k1"));
+    Assert.assertEquals(1, filler.getTop().num);
+    vq.drain("k1");
+    Assert.assertNull(filler.getTop());
+    vq.shutdown();
+  }
+
 }

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.crypto.key.kms.server;
 
 import java.io.IOException;
 import java.security.GeneralSecurityException;
+import java.security.NoSuchAlgorithmException;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
@@ -27,6 +28,7 @@ import java.util.concurrent.ExecutionException;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
 import org.apache.hadoop.crypto.key.kms.ValueQueue;
 import org.apache.hadoop.crypto.key.kms.ValueQueue.SyncGenerationPolicy;
@@ -112,6 +114,11 @@ public class EagerKeyGeneratorKeyProviderCryptoExtension
       }
     }
 
+    @Override
+    public void drain(String keyName) {
+      encKeyVersionQueue.drain(keyName);
+    }
+
     @Override
     public EncryptedKeyVersion generateEncryptedKey(String encryptionKeyName)
         throws IOException, GeneralSecurityException {
@@ -146,4 +153,19 @@ public class EagerKeyGeneratorKeyProviderCryptoExtension
         new CryptoExtension(conf, keyProviderCryptoExtension));
   }
 
+  @Override
+  public KeyVersion rollNewVersion(String name)
+      throws NoSuchAlgorithmException, IOException {
+    KeyVersion keyVersion = super.rollNewVersion(name);
+    getExtension().drain(name);
+    return keyVersion;
+  }
+
+  @Override
+  public KeyVersion rollNewVersion(String name, byte[] material)
+      throws IOException {
+    KeyVersion keyVersion = super.rollNewVersion(name, material);
+    getExtension().drain(name);
+    return keyVersion;
+  }
 }

hadoop-common-project/hadoop-kms/src/test/java/org/apache/hadoop/crypto/key/kms/server/TestKMS.java

@@ -531,6 +531,7 @@ public class TestKMS {
         Assert.assertEquals("d", meta.getDescription());
         Assert.assertEquals(attributes, meta.getAttributes());
 
+        // test delegation token retrieval
         KeyProviderDelegationTokenExtension kpdte =
             KeyProviderDelegationTokenExtension.
                 createKeyProviderDelegationTokenExtension(kp);
@@ -542,6 +543,22 @@ public class TestKMS {
 
         Assert.assertEquals(new Text("kms-dt"), credentials.getToken(
             SecurityUtil.buildTokenService(kmsAddr)).getKind());
+
+
+        // test rollover draining
+        KeyProviderCryptoExtension kpce = KeyProviderCryptoExtension.
+            createKeyProviderCryptoExtension(kp);
+        options = new KeyProvider.Options(conf);
+        options.setCipher("AES/CTR/NoPadding");
+        options.setBitLength(128);
+        kpce.createKey("k6", options);
+
+        EncryptedKeyVersion ekv1 = kpce.generateEncryptedKey("k6");
+        kpce.rollNewVersion("k6");
+        EncryptedKeyVersion ekv2 = kpce.generateEncryptedKey("k6");
+        Assert.assertNotEquals(ekv1.getEncryptionKeyVersionName(),
+            ekv2.getEncryptionKeyVersionName());
+
         return null;
       }
     });