code: public synchronized void start() {
      if (!_started.get()) {
        _executorService = new ScheduledThreadPoolExecutor(1,
            ExecutorsUtils.newThreadFactory(Optional.of(LoggerFactory.getLogger(FineGrainedWatermarkTracker.class))));
        _executorService.scheduleAtFixedRate(_sweeper, 0, _sweepIntervalMillis, TimeUnit.MILLISECONDS);
        _executorService.scheduleAtFixedRate(_stabilityChecker, 0, _stabilityCheckIntervalMillis, TimeUnit.MILLISECONDS);
      }
      _started.set(true);
    }
docs: Schedule the sweeper and stability checkers