Because it happens during startup, when no user code is running, it cannot deadlock with checkpointing.
964376
@Override
    public CompletableFuture<Void> createAllLeasesIfNotExists(List<String> partitionIds) {
        CompletableFuture<Void> future = null;

        // Optimization: list the blobs currently existing in the directory. If there are the
        // expected number of blobs, then we can skip doing the creates.
        int blobCount = 0;
        try {
            Iterable<ListBlobItem> leaseBlobs = this.consumerGroupDirectory.listBlobs("", true, null, this.leaseOperationOptions, null);
            Iterator<ListBlobItem> blobIterator = leaseBlobs.iterator();
            while (blobIterator.hasNext()) {
                blobCount++;
                blobIterator.next();
            }
        } catch (URISyntaxException | StorageException e) {
            TRACE_LOGGER.error(this.hostContext.withHost("Exception checking lease existence - leaseContainerName: " + this.storageContainerName + " consumerGroupName: "
                    + this.hostContext.getConsumerGroupName() + " storageBlobPrefix: " + this.storageBlobPrefix), e);
            future = new CompletableFuture<Void>();
            future.completeExceptionally(LoggingUtils.wrapException(e, EventProcessorHostActionStrings.CREATING_LEASES));
        }

        if (future == null) {
            // No error checking the list, so keep going
            if (blobCount == partitionIds.size()) {
                // All expected blobs found, so short-circuit
                future = CompletableFuture.completedFuture(null);
            } else {
                // Create the blobs in parallel
                ArrayList<CompletableFuture<CompleteLease>> createFutures = new ArrayList<CompletableFuture<CompleteLease>>();

                for (String id : partitionIds) {
                    CompletableFuture<CompleteLease> oneCreate = CompletableFuture.supplyAsync(() -> {
                        CompleteLease returnLease = null;
                        try {
                            returnLease = createLeaseIfNotExistsInternal(id, this.leaseOperationOptions);
                        } catch (URISyntaxException | IOException | StorageException e) {
                            TRACE_LOGGER.error(this.hostContext.withHostAndPartition(id,
                                    "Exception creating lease - leaseContainerName: " + this.storageContainerName + " consumerGroupName: " + this.hostContext.getConsumerGroupName()
                                            + " storageBlobPrefix: " + this.storageBlobPrefix), e);
                            throw LoggingUtils.wrapException(e, EventProcessorHostActionStrings.CREATING_LEASES);
                        }
                        return returnLease;
                    }, this.hostContext.getExecutor());
                    createFutures.add(oneCreate);
                }

                CompletableFuture<?>[] dummy = new CompletableFuture<?>[createFutures.size()];
                future = CompletableFuture.allOf(createFutures.toArray(dummy));
            }
        }

        return future;
    }
107009
private boolean checkVmIsStarted() {

		DescribeInstancesRequest dis = new DescribeInstancesRequest();
		dis.setInstanceIds(Collections.singletonList(this.machineId));

		DescribeInstancesResult disresult = this.ec2Api.describeInstances( dis );
		// Obtain availability zone (for later use, eg. volume attachment).
		// Necessary if no availability zone is specified in configuration
		// (because volumes must be attached to instances in the same availability zone).
		this.availabilityZone = disresult.getReservations().get(0).getInstances().get(0).getPlacement().getAvailabilityZone();
		return "running".equalsIgnoreCase( disresult.getReservations().get(0).getInstances().get(0).getState().getName());
	}
1256695
protected final void requestCheckpoint(boolean persistent)
                    throws ObjectManagerException
    {
        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.entry(this,
                        cclass,
                        "requestCheckpoint",
                        new Object[] { new Boolean(persistent) });

        // During restart checkpointHelper is not running so don't request a checkpoint.
        // A checkpoint is forced at startup once it is safe to do so.
        if (checkpointHelper != null)
            checkpointHelper.requestCheckpoint(persistent);

        if (Tracing.isAnyTracingEnabled() && trace.isEntryEnabled())
            trace.exit(this,
                       cclass,
                       "requestCheckpoint");
    }
77249
void checkNotClosed() throws SISessionUnavailableException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "checkNotClosed");

    synchronized (this)
    {
      if (_closed)
      {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.exit(tc, "checkNotClosed", "Object closed");

        throw new SISessionUnavailableException(
          nls.getFormattedMessage("OBJECT_CLOSED_ERROR_CWSIP0081", 
                                  new Object[] { _dest.getName(),
                                                 _dest.getMessageProcessor().getMessagingEngineName() }, 
                                  null));
      }
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "checkNotClosed");
  }
391838
public void lockClusterState(ClusterStateChange stateChange, Address initiator, String txnId, long leaseTime,
                                 int memberListVersion, int partitionStateVersion) {
        Preconditions.checkNotNull(stateChange);
        clusterServiceLock.lock();
        try {
            if (!node.getNodeExtension().isStartCompleted()) {
                throw new IllegalStateException("Can not lock cluster state! Startup is not completed yet!");
            }

            if (node.getClusterService().getClusterJoinManager().isMastershipClaimInProgress()) {
                throw new IllegalStateException("Can not lock cluster state! Mastership claim is in progress!");
            }

            if (stateChange.isOfType(Version.class)) {
                validateNodeCompatibleWith((Version) stateChange.getNewState());
                validateClusterVersionChange((Version) stateChange.getNewState());
            }

            checkMemberListVersion(memberListVersion);
            checkMigrationsAndPartitionStateVersion(stateChange, partitionStateVersion);

            lockOrExtendClusterState(initiator, txnId, leaseTime);

            try {
                // check migration status and partition-state version again
                // if partition state is changed then release the lock and fail.
                checkMigrationsAndPartitionStateVersion(stateChange, partitionStateVersion);
            } catch (IllegalStateException e) {
                stateLockRef.set(LockGuard.NOT_LOCKED);
                throw e;
            }
        } finally {
            clusterServiceLock.unlock();
        }
    }
489816
private void restart(long globalModVersionOfFailover) {
		try {
			if (transitionState(JobStatus.CREATED, JobStatus.RUNNING)) {
				// if we have checkpointed state, reload it into the executions
				if (executionGraph.getCheckpointCoordinator() != null) {
					// we restart the checkpoint scheduler for
					// i) enable new checkpoint could be triggered without waiting for last checkpoint expired.
					// ii) ensure the EXACTLY_ONCE semantics if needed.
					executionGraph.getCheckpointCoordinator().abortPendingCheckpoints(
						new CheckpointException(CheckpointFailureReason.JOB_FAILOVER_REGION));

					executionGraph.getCheckpointCoordinator().restoreLatestCheckpointedState(
						tasks, false, true);
				}

				HashSet<AllocationID> previousAllocationsInRegion = new HashSet<>(connectedExecutionVertexes.size());
				for (ExecutionVertex connectedExecutionVertex : connectedExecutionVertexes) {
					AllocationID latestPriorAllocation = connectedExecutionVertex.getLatestPriorAllocation();
					if (latestPriorAllocation != null) {
						previousAllocationsInRegion.add(latestPriorAllocation);
					}
				}

				//TODO, use restart strategy to schedule them.
				//restart all connected ExecutionVertexes
				for (ExecutionVertex ev : connectedExecutionVertexes) {
					try {
						ev.scheduleForExecution(
							executionGraph.getSlotProvider(),
							executionGraph.isQueuedSchedulingAllowed(),
							LocationPreferenceConstraint.ANY,
							previousAllocationsInRegion); // some inputs not belonging to the failover region might have failed concurrently
					}
					catch (Throwable e) {
						failover(globalModVersionOfFailover);
					}
				}
			}
			else {
				LOG.info("FailoverRegion {} switched from CREATED to RUNNING fail, will fail this region again.", id);
				failover(globalModVersionOfFailover);
			}
		} catch (Exception e) {
			LOG.info("FailoverRegion {} restart failed, failover again.", id, e);
			failover(globalModVersionOfFailover);
		}
	}
491396
private void checkNotClosed() throws SISessionUnavailableException 
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "checkNotClosed");

    // Check that the consumer session isn't closed
    _consumerSession.checkNotClosed();
    
    // Now check that this consumer hasn't closed.
    synchronized (this)
    {
      if(_closed)
      {
        SISessionUnavailableException e =
          new SISessionUnavailableException(
            nls.getFormattedMessage(
              "CONSUMER_CLOSED_ERROR_CWSIP0177",
              new Object[] { _localConsumerPoint.getConsumerManager().getDestination().getName(),
                             _localConsumerPoint.getConsumerManager().getMessageProcessor().getMessagingEngineName()},                
              null));

        if (tc.isEntryEnabled())
          SibTr.exit(tc, "checkNotClosed", "consumer closed");

        throw e;
      }
    }
      
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "checkNotClosed");
  }
643924
private void handleCheckpointFailure() {
    setCheckpointFailures(checkpointFailures + 1);
    if (checkpointFailures > MAX_CHECKPOINT_FAILURES) {
      LOG.fatal("Standby: Checkpointing - standby failed to checkpoint in "
          + checkpointFailures + " attempts. Aborting");
    } else {
      // We want to give some time for some transition error to recover
      // and DNS caching to expire. This is mainly for small clusters
      // where checkpoint can be very fast. Doesn't hurt if we sleep
      // on large clusters too.
      //
      LOG.info("Sleeping " + CHECKPOINT_SLEEP_BEFORE_RETRY
          + " msecs before retry checkpoints...");
      try {
        Thread.sleep(CHECKPOINT_SLEEP_BEFORE_RETRY);
        return;
      } catch (InterruptedException ie) {
        LOG.warn("Standby: Checkpointing - Thread interrupted"
            + " while sleeping before a retry.", ie);
      }
    }
    FSEditLog.runtime.exit(-1);
  }
245701
private void checkContinueWork() throws InterruptedException {
        boolean working = arbitrateEventService.mainStemEvent().check(pipelineId);
        if (!working) {
            logger.warn("mainstem is not run in this node");
            stopup(true);
        }

    }
613431
public boolean isCheckpointingEnabled() {

		if (snapshotSettings == null) {
			return false;
		}

		long checkpointInterval = snapshotSettings.getCheckpointCoordinatorConfiguration().getCheckpointInterval();
		return checkpointInterval > 0 &&
			checkpointInterval < Long.MAX_VALUE;
	}
550444
def run_checkpointing
      # task check pointer is scoped to separate class to avoid
      # the secondary thread and main thread modifying the same @task
      # variable (and accidentally clobbering it in the process)
      task_checkpointer = TaskCheckPointer.new(@task.id)
      Thread.new do
        with_thread_name("task:#{@task.id}-checkpoint") do
          while true
            sleep(Config.task_checkpoint_interval)
            task_checkpointer.checkpoint
          end
        end
      end
    end
743457
public static void assertUnlocked(Object object, String name) {
        if (RUNTIME_ASSERTIONS) {
            if (Thread.holdsLock(object)) {
                throw new RuntimeAssertion("Recursive lock of %s", name);
            }
        }
    }
273654
public boolean isSyncEnabled() {

        CmsSynchronizeSettings syncSettings = getSettings().getUserSettings().getSynchronizeSettings();
        return (syncSettings != null) && syncSettings.isSyncEnabled();
    }
559488
private void populateLockStatus(PipelineInstanceModel instanceModel, boolean isLockable, boolean isCurrentlyLocked, boolean canBeUnlocked) {
        instanceModel.setIsLockable(isLockable);
        instanceModel.setCurrentlyLocked(isCurrentlyLocked);
        instanceModel.setCanUnlock(canBeUnlocked);
    }
401598
private boolean validateProcessInstance(ProcessInstance processInst) {
        Integer status = processInst.getStatusCode();
        if (WorkStatus.STATUS_CANCELLED.equals(status)) {
            logger.info("ProcessInstance has been cancelled. ProcessInstanceId = " + processInst.getId());
            return false;
        } else if (WorkStatus.STATUS_COMPLETED.equals(status)) {
            logger.info("ProcessInstance has been completed. ProcessInstanceId = " + processInst.getId());
            return false;
        } else return true;
    }
230534
public final boolean start() {

        if (running || booting || stopping || terminated) {
            // do nothing if this instance is not stopped
            return false;
        } else {
            // mark this instance started
            booting = true;
            running = true;

            onStarted();

            return true;
        }
    }
425866
private void assertInitialized() {

		if (!services.isShuttingDown() && !services.isShutdownDone()) {

			// wait for service layer to be initialized
			while (!services.isInitialized()) {

				try { Thread.sleep(1000); } catch (Throwable t) {}
			}
		}
	}