public void close() {
    LOG.debug("Stopping backup managers for task {}.", taskName);
    stateBackendToBackupManager.values().forEach(storageBackupManager -> {
      if (storageBackupManager != null) {
        storageBackupManager.close();
      }
    });
  }
Close all the state backup managers
samza-core/src/main/java/org/apache/samza/storage/TaskStorageCommitManager.java