private void onNewJobModelConfirmed(final String nextJMVersion) {
    LOG.info("pid=" + processorId + "new version " + nextJMVersion + " of the job model got confirmed");

    String prevVersion = currentJMVersion.get();

    //Start heart-beating to new entry only when barrier reached.
    //Changing the current job model version enables that since we are heartbeating to a row identified by the current job model version.
    currentJMVersion.getAndSet(nextJMVersion);

    // Delete previous value
    ProcessorEntity entity = table.getEntity(prevVersion, processorId);
    if (entity != null) {
      entity.setEtag("*");
      table.deleteProcessorEntity(entity);
    }
    entity = table.getEntity(INITIAL_STATE, processorId);
    if (entity != null) {
      entity.setEtag("*");
      table.deleteProcessorEntity(entity);
    }

    //Start the container with the new model
    if (coordinatorListener != null) {
      coordinatorListener.onNewJobModel(processorId, jobModel);
    }
  }
Called when the JC detects that the barrier has completed by checking the barrier state on the blob.
@param nextJMVersion The new job model version after rebalancing.
samza-azure/src/main/java/org/apache/samza/coordinator/AzureJobCoordinator.java