@Override
  public ScheduledFuture scheduleTask() {
    return scheduler.scheduleWithFixedDelay(() -> {
      try {
        if (!table.getEntity(currentJMVersion.get(), processorId).getIsLeader()) {
          LOG.info("Not the leader anymore. Shutting down LeaderBarrierCompleteScheduler.");
          barrierTimeout.getAndSet(true);
          listener.onStateChange();
        } else {
          LOG.info("Leader checking for barrier state");
          // Get processor IDs listed in the table that have the new job model verion.
          Iterable<ProcessorEntity> tableList = table.getEntitiesWithPartition(nextJMVersion);
          Set<String> tableProcessors = new HashSet<>();
          for (ProcessorEntity entity : tableList) {
            tableProcessors.add(entity.getRowKey());
          }
          LOG.info("List of live processors as seen on the blob = {}", blobProcessorSet);
          LOG.info("List of live processors as seen in the table = {}", tableProcessors);
          if ((System.currentTimeMillis() - startTime) > (BARRIER_TIMEOUT_SEC * 1000)) {
            barrierTimeout.getAndSet(true);
            listener.onStateChange();
          } else if (blobProcessorSet.equals(tableProcessors)) {
            listener.onStateChange();
          }
        }
      } catch (Exception e) {
        errorHandler.accept("Exception in LeaderBarrierCompleteScheduler. Stopping the processor...");
      }
    }, BARRIER_REACHED_DELAY_SEC, BARRIER_REACHED_DELAY_SEC, TimeUnit.SECONDS);
  }
Scheduler class invoked by the leader to check if the barrier has completed.
Checks every 15 seconds.
The leader polls the Azure processor table in order to track this.
The barrier is completed if all processors that are listed alive on the blob, have entries in the Azure table with the new job model version.
All time units are in SECONDS.
samza-azure/src/main/java/org/apache/samza/coordinator/scheduler/LeaderBarrierCompleteScheduler.java