public void onBarrierStateChanged(final String version, ZkBarrierForVersionUpgrade.State state) {
      LOG.info("JobModel version " + version + " obtained consensus successfully!");
      metrics.barrierStateChange.inc();
      metrics.singleBarrierRebalancingTime.update(System.nanoTime() - startTime);
      if (ZkBarrierForVersionUpgrade.State.DONE.equals(state)) {
        debounceTimer.scheduleAfterDebounceTime(barrierAction, 0, () -> {
          LOG.info("pid=" + processorId + "new version " + version + " of the job model got confirmed");
          /*
           * Publish the active job model version separately to denote that the job model version is agreed by
           * the quorum. The active job model version is used by processors as an optimization during their startup
           * so that processors can start with the work assignment that was agreed by the quorum and allows the
           * leader to skip the rebalance if there is no change in the work assignment for the quorum across
           * quorum changes (processors leaving or joining)
           */
          if (leaderElector.amILeader()) {
            zkUtils.publishActiveJobModelVersion(version);
          }
          onNewJobModel(getJobModel());
        });
      } else {
        if (ZkBarrierForVersionUpgrade.State.TIMED_OUT.equals(state)) {
          // no-op for non-leaders
          // for leader: make sure we do not stop - so generate a new job model
          LOG.warn("Barrier for version " + version + " timed out.");
          if (leaderElector.amILeader()) {
            LOG.info("Leader will schedule a new job model generation");
            // actual actions to do are the same as onProcessorChange
            debounceTimer.scheduleAfterDebounceTime(ON_PROCESSOR_CHANGE, debounceTimeMs, ZkJobCoordinator.this::doOnProcessorChange);
          }
        }
      }
    }
Builds the {@link GrouperMetadataImpl} based upon provided {@param jobModelVersion}
and {@param processorNodes}.
@param jobModelVersion the most recent jobModelVersion available in the zookeeper.
@param processorNodes the list of live processors in the zookeeper.
@return the built grouper metadata.
samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java