@VisibleForTesting
  void startWorkWithLastActiveJobModel() {
    LOG.info("Starting the processor with the recent active job model");
    String lastActiveJobModelVersion = zkUtils.getLastActiveJobModelVersion();
    String latestJobModelVersion = zkUtils.getJobModelVersion();

    if (lastActiveJobModelVersion != null && lastActiveJobModelVersion.equals(latestJobModelVersion)) {
      final JobModel lastActiveJobModel = readJobModelFromMetadataStore(lastActiveJobModelVersion);

      /*
       * TODO: SAMZA-2645: Allow onNewJobModel as a valid state transition. Due to this limitation, we are forced
       *  to invoke onJobModelExpired even if there is nothing to expire.
       */
      checkAndExpireJobModel(lastActiveJobModel);
      onNewJobModel(lastActiveJobModel);
    }
  }
Start the processor with the last known active job model. It is safe to start with last active job model
version in all the scenarios unless in the event of concurrent rebalance. We define safe as a way to ensure that no
two processors in the quorum have overlapping work assignments.
In case of a concurrent rebalance there two scenarios
  1. Job model version update happens before processor registration
  2. Job model version update happens after processor registration
ZK guarantees FIFO order for client operations, the processor is guaranteed to see all the state up until its
own registration.
For scenario 1, due to above guarantee, the processor will not start with old assignment due to mismatch in
latest vs last active. (If there is no mismatch, the scenario reduces to one of the safe scenarios)

For scenario 2, it is possible for the processor to not see the writes by the leader about job model version change
but will eventually receive a notification on the job model version change and act on it (potentially stop
the work assignment if its not part of the job model).

In the scenario where the processor doesn't start with last active job model version, it will continue to follow
the old protocol where leader should get notified about the processor registration and potentially trigger
rebalance and notify about changes in work assignment after consensus.
TODO: SAMZA-2635: Rebalances in standalone doesn't handle DAG changes for restarted processor
samza-core/src/main/java/org/apache/samza/zk/ZkJobCoordinator.java