private void testJobModelVersionChangeHelper(JobModel activeJobModel, JobModel newJobModel,
      BiConsumer<ZkBarrierForVersionUpgrade, JobCoordinatorListener> verificationMethod) throws InterruptedException {
    final CountDownLatch completionLatch = new CountDownLatch(1);
    final JobCoordinatorListener mockListener = mock(JobCoordinatorListener.class);
    final ScheduleAfterDebounceTime mockDebounceTimer = mock(ScheduleAfterDebounceTime.class);
    final ZkBarrierForVersionUpgrade mockBarrier = mock(ZkBarrierForVersionUpgrade.class);

    doAnswer(ctx -> {
      Object[] args = ctx.getArguments();
      ((Runnable) args[2]).run();
      completionLatch.countDown();
      return null;
    }).when(mockDebounceTimer).scheduleAfterDebounceTime(anyString(), anyLong(), anyObject());


    ZkJobCoordinator zkJobCoordinator = Mockito.spy(new ZkJobCoordinator(PROCESSOR_ID, new MapConfig(),
        new NoOpMetricsRegistry(), zkUtils, zkMetadataStore, coordinatorStreamStore));
    zkJobCoordinator.setListener(mockListener);
    zkJobCoordinator.setActiveJobModel(activeJobModel);
    zkJobCoordinator.setZkBarrierUpgradeForVersion(mockBarrier);
    zkJobCoordinator.debounceTimer = mockDebounceTimer;
    doReturn(newJobModel).when(zkJobCoordinator).readJobModelFromMetadataStore(TEST_JOB_MODEL_VERSION);

    final ZkJobCoordinator.ZkJobModelVersionChangeHandler zkJobModelVersionChangeHandler =
        zkJobCoordinator.new ZkJobModelVersionChangeHandler(zkUtils);
    zkJobModelVersionChangeHandler.doHandleDataChange("path", TEST_JOB_MODEL_VERSION);
    completionLatch.await(1, TimeUnit.SECONDS);

    verificationMethod.accept(mockBarrier, mockListener);
  }
Test job model version changed without any changes to work assignment. In this scenario, existing work should
not be stopped a.k.a processor shouldn't stop the container. However, the processor proceeds to join the barrier
to notify its acceptance on the proposed job model.
samza-core/src/test/java/org/apache/samza/zk/TestZkJobCoordinator.java