@Override
  public CompletableFuture<Void> restore() {
    return CompletableFuture.runAsync(() -> {
      for (String storeName : taskStoresToRestore) {
        LOG.info("Restoring store: {} for task: {}", storeName, taskModel.getTaskName());
        SystemConsumer systemConsumer = storeConsumers.get(storeName);
        SystemStream systemStream = storeChangelogs.get(storeName);
        SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(systemStream.getSystem());
        ChangelogSSPIterator changelogSSPIterator = new ChangelogSSPIterator(systemConsumer,
            new SystemStreamPartition(systemStream, taskModel.getChangelogPartition()), null, systemAdmin, false);

        try {
          taskStores.get(storeName).restore(changelogSSPIterator);
        } catch (InterruptedException e) {
          String msg = String.format("Interrupted while restoring store: %s for task: %s",
              storeName, taskModel.getTaskName().getTaskName());
          throw new SamzaException(msg, e); // wrap in unchecked exception to throw from lambda
        }
      }
    }, restoreExecutor);
  }
Restore each store in taskStoresToRestore sequentially
samza-core/src/main/java/org/apache/samza/storage/NonTransactionalStateTaskRestoreManager.java