code: private CompletableFuture<Void> waitForCheckpointComplete(String checkpointName,
                                                                ScheduledExecutorService backgroundExecutor) {
          AtomicBoolean checkpointPending = new AtomicBoolean(true);
  
          return Futures.loop(checkpointPending::get, () -> {
              return Futures.delayedTask(() -> {
                  synchronizer.fetchUpdates();
                  checkpointPending.set(!synchronizer.getState().isCheckpointComplete(checkpointName));
                  if (checkpointPending.get()) {
                      log.debug("Waiting on checkpoint: {} currentState is: {}", checkpointName, synchronizer.getState());
                  }
                  return null;
              }, Duration.ofMillis(500), backgroundExecutor);
          }, backgroundExecutor);
      }
docs: Periodically check the state synchronizer if the given Checkpoint is complete.
  @param checkpointName Checkpoint name.
  @param backgroundExecutor Executor on which the asynchronous task will run.
  @return A CompletableFuture will be complete once the Checkpoint is complete.