@Override
  public void run() {
    // We need to get this thread so it can be interrupted if the cached proposal has been invalidated.
    _proposalPrecomputingSchedulerThread = Thread.currentThread();
    LOG.info("Starting proposal candidate computation.");
    while (!_shutdown && _numPrecomputingThreads > 0) {
      LoadMonitorTaskRunner.LoadMonitorTaskRunnerState loadMonitorTaskRunnerState = _loadMonitor.taskRunnerState();
      long sleepTimeMs = _proposalExpirationMs;
      if (loadMonitorTaskRunnerState == LOADING || loadMonitorTaskRunnerState == BOOTSTRAPPING) {
        LOG.info("Skipping proposal precomputing because load monitor is in " + loadMonitorTaskRunnerState + " state.");
        // Check in {@value HALF_MINUTE_IN_MS} to see if the load monitor state has changed.
        sleepTimeMs = HALF_MINUTE_IN_MS;
      } else if (!_loadMonitor.meetCompletenessRequirements(_requirementsWithAvailableValidWindows)) {
        LOG.info("Skipping proposal precomputing because load monitor does not have enough snapshots.");
        // Check in {@value HALF_MINUTE_IN_MS} to see if the load monitor has sufficient number of snapshots.
        sleepTimeMs = HALF_MINUTE_IN_MS;
      } else {
        try {
          if (!validCachedProposal()) {
            if (LOG.isDebugEnabled()) {
              LOG.debug("Invalidated cache. Model generation (cached: {}, current: {}).{}",
                        _cachedProposals == null ? null : _cachedProposals.modelGeneration(),
                        _loadMonitor.clusterModelGeneration(),
                        _cachedProposals == null ? "" : String.format(" Cached was excluding default topics: %s.",
                                                                      _cachedProposals.excludedTopics()));
            }
            clearCachedProposal();

            long start = System.nanoTime();
            // Proposal precomputation runs with the default topics to exclude, and allows capacity estimation.
            computeCachedProposal(_allowCapacityEstimationOnProposalPrecompute);
            _proposalComputationTimer.update(System.nanoTime() - start, TimeUnit.NANOSECONDS);
          } else {
            LOG.debug("Skipping proposal precomputing because the cached proposal result is still valid. "
                      + "Cached generation: {}", _cachedProposals.modelGeneration());
          }
        } catch (KafkaCruiseControlException e) {
          // Check in {@value HALF_MINUTE_IN_MS} to see if the ongoing execution has finished.
          sleepTimeMs = HALF_MINUTE_IN_MS;
          LOG.debug("Skipping proposal precomputing because there is an ongoing execution.", e);
        }
      }
      long deadline = _time.milliseconds() + sleepTimeMs;
      if (!_shutdown && _time.milliseconds() < deadline) {
        try {
          Thread.sleep(deadline - _time.milliseconds());
        } catch (InterruptedException e) {
          // let it go.
        }
      }
    }
  }
@Test
  public void testNoPreComputingThread() {
    GoalOptimizer goalOptimizer = createGoalOptimizer();
    // Should exit immediately.
    goalOptimizer.run();
  }