@Override
  public CompletableFuture<Void> restore() {
    return CompletableFuture.runAsync(() -> {
      Map<String, RestoreOffsets> storesToRestore = storeActions.storesToRestore;

      for (Map.Entry<String, RestoreOffsets> entry : storesToRestore.entrySet()) {
        String storeName = entry.getKey();
        String endOffset = entry.getValue().endingOffset;
        SystemStream systemStream = storeChangelogs.get(storeName);
        SystemAdmin systemAdmin = systemAdmins.getSystemAdmin(systemStream.getSystem());
        SystemConsumer systemConsumer = storeConsumers.get(storeName);
        SystemStreamPartition changelogSSP = new SystemStreamPartition(systemStream, taskModel.getChangelogPartition());

        ChangelogSSPIterator changelogSSPIterator =
            new ChangelogSSPIterator(systemConsumer, changelogSSP, endOffset, systemAdmin, true,
                currentChangelogOffsets.get(changelogSSP).getNewestOffset());
        StorageEngine taskStore = storeEngines.get(storeName);

        LOG.info("Restoring store: {} for task: {}", storeName, taskModel.getTaskName());
        try {
          taskStore.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);
  }
This class implements the state restore based on state snapshots of checkpoints and changelog.
samza-core/src/main/java/org/apache/samza/storage/TransactionalStateTaskRestoreManager.java