private SchedulerStateChangeListener createLeaderBarrierCompleteListener(String nextJMVersion, AtomicBoolean barrierTimeout) {
    return () -> {
      versionUpgradeDetected.getAndSet(false);
      String state;
      if (barrierTimeout.get()) {
        LOG.error("Barrier timed out for version {}", nextJMVersion);
        state = BarrierState.TIMEOUT.name() + " " + nextJMVersion;
      } else {
        LOG.info("Leader detected barrier completion.");
        state = BarrierState.END.name() + " " + nextJMVersion;
      }
      if (!leaderBlob.publishBarrierState(state, azureLeaderElector.getLeaseId().get())) {
        LOG.info("Leader failed to publish the job model {}. Stopping the processor with PID: .", jobModel, processorId);
        stop();
      }
      leaderBarrierScheduler.shutdown();
    };
  }
Creates a listener for LeaderBarrierCompleteScheduler class.
Invoked by the leader when it detects that rebalancing has completed by polling the processor table.
Updates the barrier state on the blob to denote that the barrier has completed.
Cancels all future tasks scheduled by the LeaderBarrierComplete scheduler to check if barrier has completed.
@return an instance of SchedulerStateChangeListener.
samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java