public void close() {
    TaskName taskName = taskModel.getTaskName();
    storeEngines.forEach((storeName, storeEngine) -> {
      if (storeEngine.getStoreProperties().isPersistedToDisk()) {
        storeEngine.stop();
        LOG.info("Stopped persistent store: {} in task: {}", storeName, taskName);
      }
    });
  }
Stop only persistent stores. In case of certain stores and store mode (such as RocksDB), this
can invoke compaction. Persisted stores are recreated in read-write mode in {@link ContainerStorageManager}.
samza-core/src/main/java/org/apache/samza/storage/TransactionalStateTaskRestoreManager.java