@Override
  public CompletableFuture<V> getAsync(K key, Object ... args) {
    incCounter(metrics.numGets);
    V value = cache.get(key, args);
    if (value != null) {
      hitCount.incrementAndGet();
      return CompletableFuture.completedFuture(value);
    }

    long startNs = clock.nanoTime();
    missCount.incrementAndGet();

    return table.getAsync(key, args).handle((result, e) -> {
      if (e != null) {
        throw new SamzaException("Failed to get the record for " + key, e);
      } else {
        if (result != null) {
          cache.put(key, result, args);
        }
        updateTimer(metrics.getNs, clock.nanoTime() - startNs);
        return result;
      }
    });
  }
Lookup the cache and return the keys that are missed in cache
@param keys keys to be looked up
@param records result map
@return list of keys missed in the cache
samza-core/src/main/java/org/apache/samza/table/caching/CachingTable.java