public void close() {

    Map<String, StorageEngine> persistentStores = this.taskStores.entrySet().stream().filter(e -> {
      return e.getValue().getStoreProperties().isPersistedToDisk();
    }).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

    persistentStores.forEach((storeName, storageEngine) -> {
      storageEngine.stop();
      this.taskStores.remove(storeName);
    });
    LOG.info("Stopped persistent stores {}", persistentStores);
  }
Stop only persistent stores. In case of certain stores and store mode (such as RocksDB), this
can invoke compaction.
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java