@Override
  public synchronized void flush() {
    incCounter(metrics.numFlushes);
    long startNs = clock.nanoTime();
    Preconditions.checkNotNull(table, "Cannot flush a read-only table: " + table);
    table.flush();
    updateTimer(metrics.flushNs, clock.nanoTime() - startNs);
  }
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