@Test
public void testNewDataEncryptionKey() throws Exception {

  final Configuration conf = new HdfsConfiguration();
  // Enable data transport encryption and access token
  conf.setBoolean(DFSConfigKeys.DFS_ENCRYPT_DATA_TRANSFER_KEY, true);
  conf.setBoolean(DFSConfigKeys.DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY, true);
  final long keyUpdateInterval = 2 * 1000;
  final long tokenLifeTime = keyUpdateInterval;
  final String blockPoolId = "bp-foo";
  FakeTimer fakeTimer = new FakeTimer();
  BlockTokenSecretManager btsm = new BlockTokenSecretManager(
      keyUpdateInterval, tokenLifeTime, 0, 1, blockPoolId, null, false);
  Whitebox.setInternalState(btsm, "timer", fakeTimer);
  // When KeyManager asks for block keys, return them from btsm directly
  NamenodeProtocol namenode = mock(NamenodeProtocol.class);
  when(namenode.getBlockKeys()).thenReturn(btsm.exportKeys());
  // Instantiate a KeyManager instance and get data encryption key.
  KeyManager keyManager = new KeyManager(blockPoolId, namenode,
      true, conf);
  Whitebox.setInternalState(keyManager, "timer", fakeTimer);
  Whitebox.setInternalState(
      Whitebox.getInternalState(keyManager, "blockTokenSecretManager"),
      "timer", fakeTimer);
  final DataEncryptionKey dek = keyManager.newDataEncryptionKey();
  final long remainingTime = dek.expiryDate - fakeTimer.now();
  assertEquals("KeyManager dataEncryptionKey should expire in 2 seconds",
      keyUpdateInterval, remainingTime);
  // advance the timer to expire the block key and data encryption key
  fakeTimer.advance(keyUpdateInterval + 1);
  // After the initial data encryption key expires, KeyManager should
  // regenerate a valid data encryption key using the current block key.
  final DataEncryptionKey dekAfterExpiration =
      keyManager.newDataEncryptionKey();
  assertNotEquals("KeyManager should generate a new data encryption key",
      dek, dekAfterExpiration);
  assertTrue("KeyManager has an expired DataEncryptionKey!",
      dekAfterExpiration.expiryDate > fakeTimer.now());
}