@VisibleForTesting
  void checkAndExpireJobModel(JobModel newJobModel) {
    Preconditions.checkNotNull(newJobModel, "JobModel cannot be null");
    if (coordinatorListener == null) {
      LOG.info("Skipping job model expiration since there are no active listeners");
      return;
    }

    LOG.info("Checking for work assignment changes for processor {} between active job model {} and new job model {}",
        processorId, activeJobModel, newJobModel);
    if (JobModelUtil.compareContainerModelForProcessor(processorId, activeJobModel, newJobModel)) {
      LOG.info("Skipping job model expiration for processor {} due to no change in work assignment.", processorId);
    } else {
      LOG.info("Work assignment changed for the processor {}. Notifying job model expiration to coordinator listener", processorId);
      coordinatorListener.onJobModelExpired();
      jobModelExpired.set(true);
    }
  }
Check if the new job model contains a different work assignment for the processor compared the last active job
model. In case of different work assignment, expire the current job model by invoking the <i>onJobModelExpired</i>
on the registered {@link JobCoordinatorListener}.
At this phase, the job model is yet to be agreed by the quorum and hence, this optimization helps availability of
the processors in the event no changes in the work assignment.

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