private void prepareWorkerExecution(JobModel jobModel, JobCoordinatorMetadata newMetadata,
      Set<JobMetadataChange> jobMetadataChanges) throws IOException {
    if (!jobMetadataChanges.isEmpty()) {
      this.jobCoordinatorMetadataManager.writeJobCoordinatorMetadata(newMetadata);
    }
    this.jobModelServingContext.setJobModel(jobModel);

    metadataResourceUtil(jobModel).createResources();

    // the fan out trigger logic comes from ClusterBasedJobCoordinator, in which a new job model can trigger a fan out
    if (this.startpointManager.isPresent() && !jobMetadataChanges.isEmpty()) {
      this.startpointManager.get().fanOut(JobModelUtil.getTaskToSystemStreamPartitions(jobModel));
    }
  }
Run set up steps so that workers can begin processing:
1. Persist job coordinator metadata
2. Publish new job model on coordinator-to-worker communication channel
3. Create metadata resources
4. Handle startpoints
samza-core/src/main/java/org/apache/samza/coordinator/staticresource/StaticResourceJobCoordinator.java