@Override
  public CompletableFuture<Map<K, V>> getAllAsync(List<K> keys, Object ... args) {
    incCounter(metrics.numGetAlls);
    // Make a copy of entries which might be immutable
    Map<K, V> getAllResult = new HashMap<>();
    List<K> missingKeys = lookupCache(keys, getAllResult);

    if (missingKeys.isEmpty()) {
      return CompletableFuture.completedFuture(getAllResult);
    }

    long startNs = clock.nanoTime();
    return table.getAllAsync(missingKeys, args).handle((records, e) -> {
      if (e != null) {
        throw new SamzaException("Failed to get records for " + keys, e);
      } else {
        if (records != null) {
          cache.putAll(records.entrySet().stream()
              .map(r -> new Entry<>(r.getKey(), r.getValue()))
              .collect(Collectors.toList()), args);
          getAllResult.putAll(records);
        }
        updateTimer(metrics.getAllNs, clock.nanoTime() - startNs);
        return getAllResult;
      }
    });
  }
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