private void onNewJobModelAvailable(final String nextJMVersion) {
    LOG.info("pid=" + processorId + "new JobModel available with job model version {}", nextJMVersion);

    //Get the new job model from blob
    jobModel = leaderBlob.getJobModel();
    LOG.info("pid=" + processorId + ": new JobModel available. ver=" + nextJMVersion + "; jm = " + jobModel);

    if (!jobModel.getContainers().containsKey(processorId)) {
      LOG.info("JobModel: {} does not contain the processorId: {}. Stopping the processor.", jobModel, processorId);
      stop();
    } else {
      //Stop current work
      if (coordinatorListener != null) {
        coordinatorListener.onJobModelExpired();
      }
      // Add entry with new job model version to the processor table
      table.addProcessorEntity(nextJMVersion, processorId, azureLeaderElector.amILeader());

      // Start polling blob to check if barrier reached
      Random random = new Random();
      String blobBarrierState = leaderBlob.getBarrierState();
      while (true) {
        if (blobBarrierState.equals(BarrierState.END.name() + " " + nextJMVersion)) {
          LOG.info("Barrier completion detected by the worker for barrier version {}.", nextJMVersion);
          versionUpgradeDetected.getAndSet(false);
          onNewJobModelConfirmed(nextJMVersion);
          break;
        } else if (blobBarrierState.equals(BarrierState.TIMEOUT.name() + " " + nextJMVersion) ||
            (Integer.valueOf(leaderBlob.getJobModelVersion()) > Integer.valueOf(nextJMVersion))) {
          LOG.info("Barrier timed out for version number {}", nextJMVersion);
          versionUpgradeDetected.getAndSet(false);
          break;
        } else {
          try {
            Thread.sleep(random.nextInt(5000));
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
          }
          LOG.info("Checking for barrier state on the blob again...");
          blobBarrierState = leaderBlob.getBarrierState();
        }
      }
    }
  }
Called when the JC detects a job model version upgrade on the shared blob.
@param nextJMVersion The new job model version after rebalancing.
samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java