@Test public void testNoPreComputingThread() { Properties props = new Properties(); props.setProperty(MonitorConfig.BOOTSTRAP_SERVERS_CONFIG, "bootstrap.servers"); props.setProperty(ExecutorConfig.ZOOKEEPER_CONNECT_CONFIG, "connect:1234"); props.setProperty(AnalyzerConfig.NUM_PROPOSAL_PRECOMPUTE_THREADS_CONFIG, "0"); props.setProperty(AnalyzerConfig.DEFAULT_GOALS_CONFIG, TestConstants.DEFAULT_GOALS_VALUES); KafkaCruiseControlConfig config = new KafkaCruiseControlConfig(props); GoalOptimizer goalOptimizer = new GoalOptimizer(config, EasyMock.mock(LoadMonitor.class), new SystemTime(), new MetricRegistry(), EasyMock.mock(Executor.class)); goalOptimizer.run(); }
@Override public void run() { _proposalPrecomputingSchedulerThread = Thread.currentThread(); LOG.info("Starting proposal candidate computation."); while (!_shutdown && _numPrecomputingThreads > 0) { LoadMonitorTaskRunner.LoadMonitorTaskRunnerState loadMonitorTaskRunnerState = _loadMonitor.taskRunnerState(); long sleepTime = _proposalExpirationMs; if (loadMonitorTaskRunnerState == LOADING || loadMonitorTaskRunnerState == BOOTSTRAPPING) { LOG.info("Skipping proposal precomputing because load monitor is in " + loadMonitorTaskRunnerState + " state."); sleepTime = 30000L; } else if (!_loadMonitor.meetCompletenessRequirements(_requirementsWithAvailableValidWindows)) { LOG.info("Skipping proposal precomputing because load monitor does not have enough snapshots."); sleepTime = 30000L; } 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(); 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) { sleepTime = 30000L; LOG.debug("Skipping proposal precomputing because there is an ongoing execution.", e); } } long deadline = _time.milliseconds() + sleepTime; if (!_shutdown && _time.milliseconds() < deadline) { try { Thread.sleep(deadline - _time.milliseconds()); } catch (InterruptedException e) { } } } }
GoalOptimizer implements Runnable { @Override public void run() { _proposalPrecomputingSchedulerThread = Thread.currentThread(); LOG.info("Starting proposal candidate computation."); while (!_shutdown && _numPrecomputingThreads > 0) { LoadMonitorTaskRunner.LoadMonitorTaskRunnerState loadMonitorTaskRunnerState = _loadMonitor.taskRunnerState(); long sleepTime = _proposalExpirationMs; if (loadMonitorTaskRunnerState == LOADING || loadMonitorTaskRunnerState == BOOTSTRAPPING) { LOG.info("Skipping proposal precomputing because load monitor is in " + loadMonitorTaskRunnerState + " state."); sleepTime = 30000L; } else if (!_loadMonitor.meetCompletenessRequirements(_requirementsWithAvailableValidWindows)) { LOG.info("Skipping proposal precomputing because load monitor does not have enough snapshots."); sleepTime = 30000L; } 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(); 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) { sleepTime = 30000L; LOG.debug("Skipping proposal precomputing because there is an ongoing execution.", e); } } long deadline = _time.milliseconds() + sleepTime; if (!_shutdown && _time.milliseconds() < deadline) { try { Thread.sleep(deadline - _time.milliseconds()); } catch (InterruptedException e) { } } } } }
GoalOptimizer implements Runnable { @Override public void run() { _proposalPrecomputingSchedulerThread = Thread.currentThread(); LOG.info("Starting proposal candidate computation."); while (!_shutdown && _numPrecomputingThreads > 0) { LoadMonitorTaskRunner.LoadMonitorTaskRunnerState loadMonitorTaskRunnerState = _loadMonitor.taskRunnerState(); long sleepTime = _proposalExpirationMs; if (loadMonitorTaskRunnerState == LOADING || loadMonitorTaskRunnerState == BOOTSTRAPPING) { LOG.info("Skipping proposal precomputing because load monitor is in " + loadMonitorTaskRunnerState + " state."); sleepTime = 30000L; } else if (!_loadMonitor.meetCompletenessRequirements(_requirementsWithAvailableValidWindows)) { LOG.info("Skipping proposal precomputing because load monitor does not have enough snapshots."); sleepTime = 30000L; } 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(); 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) { sleepTime = 30000L; LOG.debug("Skipping proposal precomputing because there is an ongoing execution.", e); } } long deadline = _time.milliseconds() + sleepTime; if (!_shutdown && _time.milliseconds() < deadline) { try { Thread.sleep(deadline - _time.milliseconds()); } catch (InterruptedException e) { } } } } GoalOptimizer(KafkaCruiseControlConfig config,
                       LoadMonitor loadMonitor,
                       Time time,
                       MetricRegistry dropwizardMetricRegistry,
                       Executor executor); }
GoalOptimizer implements Runnable { @Override public void run() { _proposalPrecomputingSchedulerThread = Thread.currentThread(); LOG.info("Starting proposal candidate computation."); while (!_shutdown && _numPrecomputingThreads > 0) { LoadMonitorTaskRunner.LoadMonitorTaskRunnerState loadMonitorTaskRunnerState = _loadMonitor.taskRunnerState(); long sleepTime = _proposalExpirationMs; if (loadMonitorTaskRunnerState == LOADING || loadMonitorTaskRunnerState == BOOTSTRAPPING) { LOG.info("Skipping proposal precomputing because load monitor is in " + loadMonitorTaskRunnerState + " state."); sleepTime = 30000L; } else if (!_loadMonitor.meetCompletenessRequirements(_requirementsWithAvailableValidWindows)) { LOG.info("Skipping proposal precomputing because load monitor does not have enough snapshots."); sleepTime = 30000L; } 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(); 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) { sleepTime = 30000L; LOG.debug("Skipping proposal precomputing because there is an ongoing execution.", e); } } long deadline = _time.milliseconds() + sleepTime; if (!_shutdown && _time.milliseconds() < deadline) { try { Thread.sleep(deadline - _time.milliseconds()); } catch (InterruptedException e) { } } } } GoalOptimizer(KafkaCruiseControlConfig config,
                       LoadMonitor loadMonitor,
                       Time time,
                       MetricRegistry dropwizardMetricRegistry,
                       Executor executor); @Override void run(); void shutdown(); ModelCompletenessRequirements defaultModelCompletenessRequirements(); ModelCompletenessRequirements modelCompletenessRequirementsForPrecomputing(); AnalyzerState state(Cluster cluster); OptimizerResult optimizations(OperationProgress operationProgress, boolean allowCapacityEstimation); OptimizerResult optimizations(ClusterModel clusterModel, OperationProgress operationProgress); OptimizerResult optimizations(ClusterModel clusterModel,
                                       List<Goal> goalsByPriority,
                                       OperationProgress operationProgress); OptimizerResult optimizations(ClusterModel clusterModel,
                                       List<Goal> goalsByPriority,
                                       OperationProgress operationProgress,
                                       Map<TopicPartition, List<ReplicaPlacementInfo>> initReplicaDistributionForProposalGeneration,
                                       OptimizationOptions optimizationOptions); Set<String> excludedTopics(ClusterModel clusterModel, Pattern requestedExcludedTopics); }
GoalOptimizer implements Runnable { @Override public void run() { _proposalPrecomputingSchedulerThread = Thread.currentThread(); LOG.info("Starting proposal candidate computation."); while (!_shutdown && _numPrecomputingThreads > 0) { LoadMonitorTaskRunner.LoadMonitorTaskRunnerState loadMonitorTaskRunnerState = _loadMonitor.taskRunnerState(); long sleepTime = _proposalExpirationMs; if (loadMonitorTaskRunnerState == LOADING || loadMonitorTaskRunnerState == BOOTSTRAPPING) { LOG.info("Skipping proposal precomputing because load monitor is in " + loadMonitorTaskRunnerState + " state."); sleepTime = 30000L; } else if (!_loadMonitor.meetCompletenessRequirements(_requirementsWithAvailableValidWindows)) { LOG.info("Skipping proposal precomputing because load monitor does not have enough snapshots."); sleepTime = 30000L; } 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(); 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) { sleepTime = 30000L; LOG.debug("Skipping proposal precomputing because there is an ongoing execution.", e); } } long deadline = _time.milliseconds() + sleepTime; if (!_shutdown && _time.milliseconds() < deadline) { try { Thread.sleep(deadline - _time.milliseconds()); } catch (InterruptedException e) { } } } } GoalOptimizer(KafkaCruiseControlConfig config,
                       LoadMonitor loadMonitor,
                       Time time,
                       MetricRegistry dropwizardMetricRegistry,
                       Executor executor); @Override void run(); void shutdown(); ModelCompletenessRequirements defaultModelCompletenessRequirements(); ModelCompletenessRequirements modelCompletenessRequirementsForPrecomputing(); AnalyzerState state(Cluster cluster); OptimizerResult optimizations(OperationProgress operationProgress, boolean allowCapacityEstimation); OptimizerResult optimizations(ClusterModel clusterModel, OperationProgress operationProgress); OptimizerResult optimizations(ClusterModel clusterModel,
                                       List<Goal> goalsByPriority,
                                       OperationProgress operationProgress); OptimizerResult optimizations(ClusterModel clusterModel,
                                       List<Goal> goalsByPriority,
                                       OperationProgress operationProgress,
                                       Map<TopicPartition, List<ReplicaPlacementInfo>> initReplicaDistributionForProposalGeneration,
                                       OptimizationOptions optimizationOptions); Set<String> excludedTopics(ClusterModel clusterModel, Pattern requestedExcludedTopics);  }