public Optional<StreamRegexMonitor> build(JobModel jobModel, Config config, StreamRegexMonitor.Callback callback) {
    JobConfig jobConfig = new JobConfig(config);

    // if input regex monitor is not enabled return empty
    if (jobConfig.getMonitorRegexDisabled()) {
      LOG.info("StreamRegexMonitor is disabled.");
      return Optional.empty();
    }

    Set<SystemStream> inputStreamsToMonitor = JobModelUtil.getSystemStreams(jobModel);
    if (inputStreamsToMonitor.isEmpty()) {
      throw new SamzaException("Input streams to a job can not be empty.");
    }

    // First list all rewriters
    Optional<String> rewritersList = jobConfig.getConfigRewriters();

    // if no rewriter is defined, there is nothing to monitor
    if (!rewritersList.isPresent()) {
      LOG.warn("No config rewriters are defined. No StreamRegexMonitor created.");
      return Optional.empty();
    }

    // Compile a map of each input-system to its corresponding input-monitor-regex patterns
    Map<String, Pattern> inputRegexesToMonitor = jobConfig.getMonitorRegexPatternMap(rewritersList.get());

    // if there are no regexes to monitor
    if (inputRegexesToMonitor.isEmpty()) {
      LOG.info("No input regexes are defined. No StreamRegexMonitor created.");
      return Optional.empty();
    }

    return Optional.of(
        new StreamRegexMonitor(inputStreamsToMonitor, inputRegexesToMonitor, this.streamMetadataCache, this.metrics,
            jobConfig.getMonitorRegexFrequency(), callback));
  }
Build a {@link StreamRegexMonitor} for input streams for the job model.
samza-core/src/main/java/org/apache/samza/coordinator/StreamRegexMonitorFactory.java