@VisibleForTesting
  void onNewJobModel(JobModel newJobModel) {
    Preconditions.checkNotNull(newJobModel, "JobModel cannot be null. Failing onNewJobModel");
    // start the container with the new model
    if (jobModelExpired.compareAndSet(true, false)) {
      LOG.info("Work assignment changed for the processor {}. Updating task locality and notifying coordinator listener", processorId);
      if (newJobModel.getContainers().containsKey(processorId)) {
        for (TaskName taskName : JobModelUtil.getTaskNamesForProcessor(processorId, newJobModel)) {
          zkUtils.writeTaskLocality(taskName, locationId);
        }

        if (coordinatorListener != null) {
          coordinatorListener.onNewJobModel(processorId, newJobModel);
        }
      }
    } else {
      /*
       * We don't expire the job model if the proposed work assignment is same as the current work assignment.
       * The implication of work assignment remaining the same can be categorized into
       *   1. Processor part of the job model
       *   2. Processor not part of the job model.
       * For both the state of the processor remains what it was when the rebalance started. e.g.,
       *   [1] should continue to process its work assignment without any interruption as part of the rebalance. i.e.,
       *       there will be no expiration of the existing work (a.k.a samza container won't be stopped) and also no
       *       notification to StreamProcessor about the rebalance since work assignment didn't change.
       *   [2] should have no work and be idle processor and will continue to be idle.
       */
      LOG.info("Skipping onNewJobModel since there are no changes in work assignment.");
    }

    /*
     * Update the last active job model to new job model regardless of whether the work assignment for the processor
     * has changed or not. It is important to do it so that all the processors has a consistent view what the latest
     * active job model is.
     */
    activeJobModel = newJobModel;
  }
Checks if the new job model contains a different work assignment for the processor compared to the last active
job model. In case of different work assignment, update the task locality of the tasks associated with the
processor and notify new job model to the registered {@link JobCoordinatorListener}.

@param newJobModel new job model agreed by the quorum
samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java