@Test
  public void testStreamRegexChange() throws IOException {
    Config jobModelConfig = mock(Config.class);
    JobModel jobModel = setupJobModel(jobModelConfig);
    StreamPartitionCountMonitor streamPartitionCountMonitor = setupStreamPartitionCountMonitor(jobModelConfig);
    StreamRegexMonitor streamRegexMonitor = mock(StreamRegexMonitor.class);
    ArgumentCaptor<StreamRegexMonitor.Callback> callbackArgumentCaptor =
        ArgumentCaptor.forClass(StreamRegexMonitor.Callback.class);
    when(this.streamRegexMonitorFactory.build(eq(jobModel), eq(jobModelConfig),
        callbackArgumentCaptor.capture())).thenReturn(Optional.of(streamRegexMonitor));
    JobCoordinatorMetadata newMetadata = setupJobCoordinatorMetadata(jobModel, jobModelConfig,
        ImmutableSet.of(JobMetadataChange.NEW_DEPLOYMENT, JobMetadataChange.JOB_MODEL), true);
    setUpDiagnosticsManager(jobModel);
    MetadataResourceUtil metadataResourceUtil = metadataResourceUtil(jobModel);
    this.staticResourceJobCoordinator.start();
    verifyStartLifecycle();
    verify(this.staticResourceJobCoordinator).doSetLoggingContextConfig(jobModelConfig);
    verify(this.diagnosticsManager).start();
    verifyPrepareWorkerExecutionAndMonitor(jobModel, metadataResourceUtil, streamPartitionCountMonitor,
        streamRegexMonitor, newMetadata, SINGLE_SSP_FANOUT);
    // call the callback from the monitor
    callbackArgumentCaptor.getValue()
        .onInputStreamsChanged(ImmutableSet.of(SYSTEM_STREAM),
            ImmutableSet.of(SYSTEM_STREAM, new SystemStream("system", "stream1")),
            ImmutableMap.of("system", Pattern.compile("stream.*")));
    verify(this.jobRestartSignal).restartJob();
  }
Missing {@link StartpointManager}, {@link JobCoordinatorListener}, {@link StreamRegexMonitor}
samza-core/src/test/java/org/apache/samza/coordinator/staticresource/TestStaticResourceJobCoordinator.java