@Override
  public void close() {
    super.close();
    tables.forEach(t -> t.close());
    rateLimitingExecutors.values().forEach(e -> e.shutdown());
    rateLimitingExecutors.clear();
    callbackExecutors.values().forEach(e -> e.shutdown());
    callbackExecutors.clear();
    batchExecutors.values().forEach(e -> e.shutdown());
    batchExecutors.clear();
  }
Map of tableId -> executor service for async table IO and callbacks. The same executors
are shared by both read/write operations such that tables of the same tableId all share
the set same of executors globally whereas table itself is per-task.
samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableProvider.java