public void start() {
    try {
      Map<String, MonitorConfig> monitorConfigs = getMonitorConfigs(config);
      for (Map.Entry<String, MonitorConfig> entry : monitorConfigs.entrySet()) {
        String monitorName = entry.getKey();
        MonitorConfig monitorConfig = entry.getValue();

        if (!Strings.isNullOrEmpty(monitorConfig.getMonitorFactoryClass())) {
          int schedulingIntervalInMs = monitorConfig.getSchedulingIntervalInMs();
          int monitorSchedulingJitterInMs = (int) (RANDOM.nextInt(schedulingIntervalInMs + 1) * (monitorConfig.getSchedulingJitterPercent() / 100.0));
          schedulingIntervalInMs += monitorSchedulingJitterInMs;
          LOGGER.info("Scheduling the monitor: {} to run every {} ms.", monitorName, schedulingIntervalInMs);
          // Create a new SchedulerExecutorService for each monitor. This ensures that a long running monitor service
          // does not block another monitor from scheduling/running. A long running monitor will not create a backlog
          // of work for future execution of the same monitor. A new monitor is scheduled only when current work is complete.
          createSchedulerAndScheduleMonitor(monitorName, monitorConfig, schedulingIntervalInMs);
        } else {
          // When MonitorFactoryClass is not defined in the config, ignore the monitor config
          LOGGER.warn("Not scheduling the monitor: {} to run, since monitor factory class is not set in config.", monitorName);
        }
      }
    } catch (InstantiationException e) {
      LOGGER.error("Exception when instantiating the monitor : ", e);
      throw new SamzaException(e);
    }
  }
The class responsible for handling long-running/scheduled monitors in Samza REST.
Takes a SamzaRestConfig object in the constructor and handles instantiation of
monitors and scheduling them to run based on the properties in the config.
samza-rest/src/main/java/org/apache/samza/monitor/SamzaMonitorService.java