HFileReader implements Reader { @Override public Future<Map<ByteBuffer, Pair<ByteBuffer, Throwable>>> getValues( List<ByteBuffer> keyList) throws Throwable { List<Future<Pair<ByteBuffer, Pair<ByteBuffer, Throwable>>>> futureList = Lists.newArrayListWithCapacity(keyList.size()); for (final ByteBuffer key : keyList) { Future<Pair<ByteBuffer, Pair<ByteBuffer, Throwable>>> future = futurePool.apply( new Function0<Pair<ByteBuffer, Pair<ByteBuffer, Throwable>>>() { @Override public Pair<ByteBuffer, Pair<ByteBuffer, Throwable>> apply() { long startTimeMillis = System.currentTimeMillis(); Pair<ByteBuffer, Pair<ByteBuffer, Throwable>> keyExceptionPair = getValueFromHFile(key); int timeMs = (int)(System.currentTimeMillis() - startTimeMillis); if (metricsStatsKey != null) { Stats.addMetric(metricsStatsKey, timeMs); } Stats.addMetric("lookup-latency-ms", timeMs); if (keyExceptionPair.getRight().getRight() != null) { if (errorStatsKey != null) { Stats.incr(errorStatsKey); } Stats.incr("lookup-errors"); } return keyExceptionPair; } }); futureList.add(future); } return Future.collect(futureList).map( new Function<List<Pair<ByteBuffer, Pair<ByteBuffer, Throwable>>>, Map<ByteBuffer, Pair<ByteBuffer, Throwable>>>() { @Override public Map<ByteBuffer, Pair<ByteBuffer, Throwable>> apply( List<Pair<ByteBuffer, Pair<ByteBuffer, Throwable>>> keyValuePairList) { Map<ByteBuffer, Pair<ByteBuffer, Throwable>> returnMap = Maps.newHashMapWithExpectedSize(keyValuePairList.size()); for (Pair<ByteBuffer, Pair<ByteBuffer, Throwable>> pair : keyValuePairList) { Pair<ByteBuffer, Throwable> value = pair.getRight(); if (value.getLeft() != null || value.getRight() != null) { returnMap.put(pair.getLeft(), pair.getRight()); } } return returnMap; } }); } HFileReader(FileSystem fs,
                     String path,
                     CacheConfig cacheConf,
                     FuturePool futurePool); @Override Future<Map<ByteBuffer, Pair<ByteBuffer, Throwable>>> getValues(
      List<ByteBuffer> keyList); @Override void close();  }
@Test public void testGetValues() throws Throwable { hfileReader.setFileSet(null); List<ByteBuffer> keyList = Lists.newArrayListWithCapacity(5000); for (int i = 0; i < 10000; i += 2) { keyList.add(ByteBuffer.wrap(String.format("%04d", i).getBytes())); } keyList.add(ByteBuffer.wrap(String.format("%04d", 15000).getBytes())); checkKeyValues(keyList, 5000, 1000, 1); assertEquals(5001, Stats.getMetric("lookup-latency-ms").apply().count()); assertEquals(1000, Stats.getCounter("lookup-errors").apply()); assertEquals(1, Stats.getCounter("not-found").apply()); hfileReader.setFileSet("test"); keyList = Lists.newArrayListWithCapacity(5000); for (int i = 1; i < 10000; i += 2) { keyList.add(ByteBuffer.wrap(String.format("%04d", i).getBytes())); } keyList.add(ByteBuffer.wrap(String.format("%04d", 16000).getBytes())); checkKeyValues(keyList, 5000, 1000, 1); assertEquals(10002, Stats.getMetric("lookup-latency-ms").apply().count()); assertEquals(5001, Stats.getMetric("test-lookup-latency-ms").apply().count()); assertEquals(8000, Stats.getMetric("value-size").apply().count()); assertEquals(4000, Stats.getMetric("test-value-size").apply().count()); assertEquals(2000, Stats.getCounter("lookup-errors").apply()); assertEquals(1000, Stats.getCounter("test-lookup-errors").apply()); assertEquals(2, Stats.getCounter("not-found").apply()); assertEquals(1, Stats.getCounter("test-not-found").apply()); }