@Override
  public Metadata getMetadata(String name) throws IOException {
    try {
      return getExtension().keyMetadataCache.get(name);
    } catch (ExecutionException ex) {
      Throwable cause = ex.getCause();
      if (cause instanceof KeyNotFoundException) {
        return null;
      } else if (cause instanceof IOException) {
        throw (IOException) cause;
      } else {
        throw new IOException(cause);
      }
    }
  }
@Test
  public void testMetadata() throws Exception {
    KeyProvider.Metadata mockMeta = Mockito.mock(KeyProvider.Metadata.class);
    KeyProvider mockProv = Mockito.mock(KeyProvider.class);
    Mockito.when(mockProv.getMetadata(Mockito.eq("k1"))).thenReturn(mockMeta);
    Mockito.when(mockProv.getMetadata(Mockito.eq("k2"))).thenReturn(null);
    Mockito.when(mockProv.getConf()).thenReturn(new Configuration());
    KeyProvider cache = new CachingKeyProvider(mockProv, 100, 100);

    // asserting caching
    Assert.assertEquals(mockMeta, cache.getMetadata("k1"));
    Mockito.verify(mockProv, Mockito.times(1)).getMetadata(Mockito.eq("k1"));
    Assert.assertEquals(mockMeta, cache.getMetadata("k1"));
    Mockito.verify(mockProv, Mockito.times(1)).getMetadata(Mockito.eq("k1"));
    Thread.sleep(200);
    Assert.assertEquals(mockMeta, cache.getMetadata("k1"));
    Mockito.verify(mockProv, Mockito.times(2)).getMetadata(Mockito.eq("k1"));

    // asserting no caching when key is not known
    cache = new CachingKeyProvider(mockProv, 100, 100);
    Assert.assertEquals(null, cache.getMetadata("k2"));
    Mockito.verify(mockProv, Mockito.times(1)).getMetadata(Mockito.eq("k2"));
    Assert.assertEquals(null, cache.getMetadata("k2"));
    Mockito.verify(mockProv, Mockito.times(2)).getMetadata(Mockito.eq("k2"));
  }