public Set<JobMetadataChange> checkForMetadataChanges(JobCoordinatorMetadata newMetadata,
      JobCoordinatorMetadata previousMetadata) {
    Set<JobMetadataChange> changes = new HashSet<>();
    boolean newDeployment = false;
    if (previousMetadata == null || !previousMetadata.getEpochId().equals(newMetadata.getEpochId())) {
      metrics.setNewDeployment(1);
      newDeployment = true;
      changes.add(JobMetadataChange.NEW_DEPLOYMENT);
    }
    if (previousMetadata == null || !previousMetadata.getJobModelId().equals(newMetadata.getJobModelId())) {
      if (!newDeployment) {
        metrics.setJobModelChangedAcrossApplicationAttempt(1);
      }
      changes.add(JobMetadataChange.JOB_MODEL);
    }
    if (previousMetadata == null || !previousMetadata.getConfigId().equals(newMetadata.getConfigId())) {
      if (!newDeployment) {
        metrics.setConfigChangedAcrossApplicationAttempt(1);
      }
      changes.add(JobMetadataChange.CONFIG);
    }
    if (changes.isEmpty()) {
      LOG.info("Job coordinator metadata {} unchanged.", newMetadata);
      metrics.incrementApplicationAttemptCount();
    } else {
      LOG.info("Job coordinator metadata changed from: {} to: {}", previousMetadata, newMetadata);
    }
    return changes;
  }
Check for changes between the metadata passed as inputs. Metadata is considered changed if any of the attributes within
{@linkplain JobCoordinatorMetadata} changes.

We intentionally check for each changes to help us track at this granularity. We want to use this information
to determine if complex handling is required to cater these changes instead of blindly restarting all the
containers upstream.

@param newMetadata new metadata to be compared
@param previousMetadata previous metadata to be compared against

@return true if metadata changed, false otherwise
samza-core/src/main/java/org/apache/samza/job/metadata/JobCoordinatorMetadataManager.java