@Override
  public CompletableFuture<Void> putAsync(K key, V value, Object ... args) {
    incCounter(metrics.numPuts);
    Preconditions.checkNotNull(table, "Cannot write to a read-only table: " + table);

    long startNs = clock.nanoTime();
    return table.putAsync(key, value, args).handle((result, e) -> {
      if (e != null) {
        throw new SamzaException("Failed to put a record, key=" + key + ", value=" + value, e);
      } else if (!isWriteAround) {
        if (value == null) {
          cache.delete(key, args);
        } else {
          cache.put(key, value, args);
        }
      }
      updateTimer(metrics.putNs, 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