@Override
  public <T> CompletableFuture<T> writeAsync(int opId, Object... args) {
    incCounter(metrics.numWrites);
    long startNs = clock.nanoTime();
    return table.writeAsync(opId, args).handle((result, e) -> {
      if (e != null) {
        throw new SamzaException("Failed to write, opId=" + opId, e);
      }
      updateTimer(metrics.writeNs, clock.nanoTime() - startNs);
      return (T) 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