@Override
  public CompletableFuture<Void> putAllAsync(List<Entry<K, V>> entries, Object ... args) {
    CompletableFuture<Void> future = new CompletableFuture<>();
    try {
      // Separate out put vs delete records
      List<K> delKeys = new ArrayList<>();
      List<Entry<K, V>> putRecords = new ArrayList<>();
      entries.forEach(r -> {
        if (r.getValue() != null) {
          putRecords.add(r);
        } else {
          delKeys.add(r.getKey());
        }
      });

      cache.invalidateAll(delKeys);
      putRecords.forEach(e -> put(e.getKey(), e.getValue()));
      future.complete(null);
    } catch (Exception e) {
      future.completeExceptionally(e);
    }
    return future;
  }
Simple cache table backed by a Guava cache instance. Application is expect to build
a cache instance with desired parameters and specify it to the table descriptor.

@param <K> type of the key in the cache
@param <V> type of the value in the cache
samza-core/src/main/java/org/apache/samza/table/caching/guava/GuavaCacheTable.java