@Test
  public void testNoExistingJobModel() throws IOException {
    Config jobModelConfig = mock(Config.class);
    JobModel jobModel = setupJobModel(jobModelConfig);
    StreamPartitionCountMonitor streamPartitionCountMonitor = setupStreamPartitionCountMonitor(jobModelConfig);
    StreamRegexMonitor streamRegexMonitor = setupStreamRegexMonitor(jobModel, jobModelConfig);
    JobCoordinatorMetadata newMetadata = setupJobCoordinatorMetadata(jobModel, jobModelConfig,
        ImmutableSet.copyOf(Arrays.asList(JobMetadataChange.values())), false);
    setUpDiagnosticsManager(jobModel);
    MetadataResourceUtil metadataResourceUtil = metadataResourceUtil(jobModel);
    this.staticResourceJobCoordinator.start();
    assertEquals(jobModel, this.staticResourceJobCoordinator.getJobModel());
    verifyStartLifecycle();
    verify(this.staticResourceJobCoordinator).doSetLoggingContextConfig(jobModelConfig);
    verify(this.diagnosticsManager).start();
    verifyPrepareWorkerExecutionAndMonitor(jobModel, metadataResourceUtil, streamPartitionCountMonitor,
        streamRegexMonitor, newMetadata, SINGLE_SSP_FANOUT);
    verify(this.jobCoordinatorListener).onNewJobModel(PROCESSOR_ID, jobModel);
  }
In general, these tests start the {@link StaticResourceJobCoordinator} in a separate thread and then execute certain
actions (trigger callbacks which will change the job model, trigger shutdown) to check the coordination flow.
samza-core/src/test/java/org/apache/samza/coordinator/staticresource/TestStaticResourceJobCoordinator.java