@VisibleForTesting
  void generateAndUpdateJobCoordinatorMetadata(JobModel jobModel) {
    JobCoordinatorMetadataManager jobCoordinatorMetadataManager = createJobCoordinatorMetadataManager();

    JobCoordinatorMetadata previousMetadata = jobCoordinatorMetadataManager.readJobCoordinatorMetadata();
    JobCoordinatorMetadata newMetadata = jobCoordinatorMetadataManager.generateJobCoordinatorMetadata(jobModel, config);
    if (!jobCoordinatorMetadataManager.checkForMetadataChanges(newMetadata, previousMetadata).isEmpty()) {
      jobCoordinatorMetadataManager.writeJobCoordinatorMetadata(newMetadata);
      metadataChangedAcrossAttempts = true;
    }
  }
Generate the job coordinator metadata for current application attempt and checks for changes in the
metadata from the previous attempt and writes the updates metadata to coordinator stream.

@param jobModel job model used to generate the job coordinator metadata
samza-core/src/main/java/org/apache/samza/clustermanager/ClusterBasedJobCoordinator.java