@Override
  public ScheduledFuture scheduleTask() {
    return scheduler.scheduleWithFixedDelay(() -> {
      try {
        if (!table.getEntity(currentJMVersion.get(), processorId).getIsLeader()) {
          LOG.info("Not the leader anymore. Shutting down LivenessCheckScheduler.");
          scheduler.shutdownNow();
          return;
        }
        LOG.info("Checking for list of live processors");
        //Get the list of live processors published on the blob.
        Set<String> currProcessors = new HashSet<>(blob.getLiveProcessorList());
        //Get the list of live processors from the table. This is the current system state.
        Set<String> liveProcessors = table.getActiveProcessorsList(currentJMVersion);
        //Invoke listener if the table list is not consistent with the blob list.
        if (!liveProcessors.equals(currProcessors)) {
          liveProcessorsList.getAndSet(new ArrayList<>(liveProcessors));
          listener.onStateChange();
        }
      } catch (Exception e) {
        errorHandler.accept("Exception in Liveness Check Scheduler. Stopping the processor...");
      }
    }, LIVENESS_CHECK_DELAY_SEC, LIVENESS_CHECK_DELAY_SEC, TimeUnit.SECONDS);
  }
Scheduler class invoked by the leader to check for changes in the list of live processors.
Checks every 30 seconds.
If a processor row hasn't been updated since 30 seconds, the system assumes that the processor has died.
All time units are in SECONDS.
samza-azure/src/main/java/org/apache/samza/coordinator/scheduler/LivenessCheckScheduler.java