@Test public void testSanityCheckDryRun() throws InterruptedException, ExecutionException, TimeoutException { KafkaCruiseControlConfig config = EasyMock.mock(KafkaCruiseControlConfig.class); Time time = EasyMock.mock(Time.class); AnomalyDetectorManager anomalyDetectorManager = EasyMock.mock(AnomalyDetectorManager.class); Executor executor = EasyMock.strictMock(Executor.class); LoadMonitor loadMonitor = EasyMock.mock(LoadMonitor.class); ExecutorService goalOptimizerExecutor = EasyMock.mock(ExecutorService.class); GoalOptimizer goalOptimizer = EasyMock.mock(GoalOptimizer.class); EasyMock.expect(executor.hasOngoingExecution()).andReturn(true).times(2); EasyMock.expect(executor.hasOngoingExecution()).andReturn(false).once(); EasyMock.expect(executor.hasOngoingPartitionReassignments()).andReturn(true); EasyMock.expect(executor.hasOngoingExecution()).andReturn(false).once(); EasyMock.expect(executor.hasOngoingPartitionReassignments()).andReturn(false); EasyMock.expect(executor.hasOngoingLeaderElection()).andReturn(true); EasyMock.expect(executor.hasOngoingExecution()).andReturn(false).once(); EasyMock.expect(executor.hasOngoingPartitionReassignments()).andReturn(false); EasyMock.expect(executor.hasOngoingLeaderElection()).andReturn(false); EasyMock.replay(config, time, anomalyDetectorManager, executor, loadMonitor, goalOptimizerExecutor, goalOptimizer); KafkaCruiseControl kafkaCruiseControl = new KafkaCruiseControl(config, time, anomalyDetectorManager, executor, loadMonitor, goalOptimizerExecutor, goalOptimizer); kafkaCruiseControl.sanityCheckDryRun(true, false); kafkaCruiseControl.sanityCheckDryRun(true, true); kafkaCruiseControl.sanityCheckDryRun(false, true); assertThrows(IllegalStateException.class, () -> kafkaCruiseControl.sanityCheckDryRun(false, false)); assertThrows(IllegalStateException.class, () -> kafkaCruiseControl.sanityCheckDryRun(false, false)); assertThrows(IllegalStateException.class, () -> kafkaCruiseControl.sanityCheckDryRun(false, false)); kafkaCruiseControl.sanityCheckDryRun(false, false); EasyMock.verify(config, time, anomalyDetectorManager, executor, loadMonitor, goalOptimizerExecutor, goalOptimizer); }
public void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution) { if (dryRun) { return; } if (hasOngoingExecution()) { if (!stopOngoingExecution) { throw new IllegalStateException(String.format("Cannot start a new execution while there is an ongoing execution. " + "Please use %s=true to stop ongoing execution and start a new one.", STOP_ONGOING_EXECUTION_PARAM)); } } else { if (_executor.hasOngoingPartitionReassignments()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing partition reassignments initiated by " + "external agent."); } else if (_executor.hasOngoingLeaderElection()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing leadership reassignments initiated by " + "external agent."); } } }
KafkaCruiseControl { public void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution) { if (dryRun) { return; } if (hasOngoingExecution()) { if (!stopOngoingExecution) { throw new IllegalStateException(String.format("Cannot start a new execution while there is an ongoing execution. " + "Please use %s=true to stop ongoing execution and start a new one.", STOP_ONGOING_EXECUTION_PARAM)); } } else { if (_executor.hasOngoingPartitionReassignments()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing partition reassignments initiated by " + "external agent."); } else if (_executor.hasOngoingLeaderElection()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing leadership reassignments initiated by " + "external agent."); } } } }
KafkaCruiseControl { public void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution) { if (dryRun) { return; } if (hasOngoingExecution()) { if (!stopOngoingExecution) { throw new IllegalStateException(String.format("Cannot start a new execution while there is an ongoing execution. " + "Please use %s=true to stop ongoing execution and start a new one.", STOP_ONGOING_EXECUTION_PARAM)); } } else { if (_executor.hasOngoingPartitionReassignments()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing partition reassignments initiated by " + "external agent."); } else if (_executor.hasOngoingLeaderElection()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing leadership reassignments initiated by " + "external agent."); } } } KafkaCruiseControl(KafkaCruiseControlConfig config, MetricRegistry dropwizardMetricRegistry);   KafkaCruiseControl(KafkaCruiseControlConfig config,
                     Time time,
                     AnomalyDetectorManager anomalyDetectorManager,
                     Executor executor,
                     LoadMonitor loadMonitor,
                     ExecutorService goalOptimizerExecutor,
                     GoalOptimizer goalOptimizer); }
KafkaCruiseControl { public void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution) { if (dryRun) { return; } if (hasOngoingExecution()) { if (!stopOngoingExecution) { throw new IllegalStateException(String.format("Cannot start a new execution while there is an ongoing execution. " + "Please use %s=true to stop ongoing execution and start a new one.", STOP_ONGOING_EXECUTION_PARAM)); } } else { if (_executor.hasOngoingPartitionReassignments()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing partition reassignments initiated by " + "external agent."); } else if (_executor.hasOngoingLeaderElection()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing leadership reassignments initiated by " + "external agent."); } } } KafkaCruiseControl(KafkaCruiseControlConfig config, MetricRegistry dropwizardMetricRegistry);   KafkaCruiseControl(KafkaCruiseControlConfig config,
                     Time time,
                     AnomalyDetectorManager anomalyDetectorManager,
                     Executor executor,
                     LoadMonitor loadMonitor,
                     ExecutorService goalOptimizerExecutor,
                     GoalOptimizer goalOptimizer); LoadMonitor loadMonitor(); MetadataClient.ClusterAndGeneration refreshClusterAndGeneration(); LoadMonitorTaskRunner.LoadMonitorTaskRunnerState getLoadMonitorTaskRunnerState(); LoadMonitor.AutoCloseableSemaphore acquireForModelGeneration(OperationProgress operationProgress); long timeMs(); void sleep(long ms); void startUp(); void shutdown(); void setUserTaskManagerInExecutor(UserTaskManager userTaskManager); void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution); boolean hasOngoingExecution(); boolean modifyOngoingExecution(boolean modify); BrokerStats cachedBrokerLoadStats(boolean allowCapacityEstimation); ClusterModel clusterModel(ModelCompletenessRequirements requirements,
                                   boolean allowCapacityEstimation,
                                   OperationProgress operationProgress); ClusterModel clusterModel(long from,
                                   long to,
                                   ModelCompletenessRequirements requirements,
                                   boolean populateReplicaPlacementInfo,
                                   boolean allowCapacityEstimation,
                                   OperationProgress operationProgress); ClusterModel clusterCapacity(); void bootstrap(Long startMs, Long endMs, boolean clearMetrics); void pauseMetricSampling(String reason); void train(Long startMs, Long endMs); boolean setSelfHealingFor(AnomalyType anomalyType, boolean isSelfHealingEnabled); boolean setConcurrencyAdjusterFor(ConcurrencyType concurrencyType, boolean isConcurrencyAdjusterEnabled); boolean dropRecentBrokers(Set<Integer> brokersToDrop, boolean isRemoved); void addRecentBrokersPermanently(Set<Integer> brokersToAdd, boolean isRemoved); Set<Integer> recentBrokers(boolean isRemoved); void setRequestedExecutionProgressCheckIntervalMs(Long requestedExecutionProgressCheckIntervalMs); void setRequestedInterBrokerPartitionMovementConcurrency(Integer requestedInterBrokerPartitionMovementConcurrency); void setRequestedIntraBrokerPartitionMovementConcurrency(Integer requestedIntraBrokerPartitionMovementConcurrency); void setRequestedLeadershipMovementConcurrency(Integer requestedLeadershipMovementConcurrency); void resumeMetricSampling(String reason); OptimizerResult getProposals(OperationProgress operationProgress, boolean allowCapacityEstimation); boolean ignoreProposalCache(List<String> goals,
                                     ModelCompletenessRequirements requirements,
                                     Pattern excludedTopics,
                                     boolean excludeBrokers,
                                     boolean ignoreProposalCache,
                                     boolean isTriggeredByGoalViolation,
                                     Set<Integer> requestedDestinationBrokerIds,
                                     boolean isRebalanceDiskMode); synchronized OptimizerResult optimizations(ClusterModel clusterModel,
                                                    List<Goal> goalsByPriority,
                                                    OperationProgress operationProgress,
                                                    Map<TopicPartition, List<ReplicaPlacementInfo>> initReplicaDistribution,
                                                    OptimizationOptions optimizationOptions); Set<String> excludedTopics(ClusterModel clusterModel, Pattern requestedExcludedTopics); KafkaCruiseControlConfig config(); void executeProposals(Set<ExecutionProposal> proposals,
                               Set<Integer> unthrottledBrokers,
                               boolean isKafkaAssignerMode,
                               Integer concurrentInterBrokerPartitionMovements,
                               Integer concurrentIntraBrokerPartitionMovements,
                               Integer concurrentLeaderMovements,
                               Long executionProgressCheckIntervalMs,
                               ReplicaMovementStrategy replicaMovementStrategy,
                               Long replicationThrottle,
                               boolean isTriggeredByUserRequest,
                               String uuid,
                               boolean skipAutoRefreshingConcurrency); void executeRemoval(Set<ExecutionProposal> proposals,
                             boolean throttleDecommissionedBroker,
                             Set<Integer> removedBrokers,
                             boolean isKafkaAssignerMode,
                             Integer concurrentInterBrokerPartitionMovements,
                             Integer concurrentLeaderMovements,
                             Long executionProgressCheckIntervalMs,
                             ReplicaMovementStrategy replicaMovementStrategy,
                             Long replicationThrottle,
                             boolean isTriggeredByUserRequest,
                             String uuid); void executeDemotion(Set<ExecutionProposal> proposals,
                              Set<Integer> demotedBrokers,
                              Integer concurrentLeaderMovements,
                              int brokerCount,
                              Long executionProgressCheckIntervalMs,
                              ReplicaMovementStrategy replicaMovementStrategy,
                              Long replicationThrottle,
                              boolean isTriggeredByUserRequest,
                              String uuid); void userTriggeredStopExecution(boolean forceExecutionStop); void setGeneratingProposalsForExecution(String uuid, Supplier<String> reasonSupplier, boolean isTriggeredByUserRequest); synchronized void failGeneratingProposalsForExecution(String uuid); long executionProgressCheckIntervalMs(); ExecutorState.State executionState(); ExecutorState executorState(); LoadMonitorState monitorState(Cluster cluster); AnalyzerState analyzerState(Cluster cluster); AnomalyDetectorState anomalyDetectorState(); Cluster kafkaCluster(); TopicConfigProvider topicConfigProvider(); static String cruiseControlVersion(); static String cruiseControlCommitId(); ModelCompletenessRequirements modelCompletenessRequirements(Collection<Goal> goals); boolean meetCompletenessRequirements(List<String> goals); void sanityCheckBrokerPresence(Set<Integer> brokerIds); }
KafkaCruiseControl { public void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution) { if (dryRun) { return; } if (hasOngoingExecution()) { if (!stopOngoingExecution) { throw new IllegalStateException(String.format("Cannot start a new execution while there is an ongoing execution. " + "Please use %s=true to stop ongoing execution and start a new one.", STOP_ONGOING_EXECUTION_PARAM)); } } else { if (_executor.hasOngoingPartitionReassignments()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing partition reassignments initiated by " + "external agent."); } else if (_executor.hasOngoingLeaderElection()) { throw new IllegalStateException("Cannot execute new proposals while there are ongoing leadership reassignments initiated by " + "external agent."); } } } KafkaCruiseControl(KafkaCruiseControlConfig config, MetricRegistry dropwizardMetricRegistry);   KafkaCruiseControl(KafkaCruiseControlConfig config,
                     Time time,
                     AnomalyDetectorManager anomalyDetectorManager,
                     Executor executor,
                     LoadMonitor loadMonitor,
                     ExecutorService goalOptimizerExecutor,
                     GoalOptimizer goalOptimizer); LoadMonitor loadMonitor(); MetadataClient.ClusterAndGeneration refreshClusterAndGeneration(); LoadMonitorTaskRunner.LoadMonitorTaskRunnerState getLoadMonitorTaskRunnerState(); LoadMonitor.AutoCloseableSemaphore acquireForModelGeneration(OperationProgress operationProgress); long timeMs(); void sleep(long ms); void startUp(); void shutdown(); void setUserTaskManagerInExecutor(UserTaskManager userTaskManager); void sanityCheckDryRun(boolean dryRun, boolean stopOngoingExecution); boolean hasOngoingExecution(); boolean modifyOngoingExecution(boolean modify); BrokerStats cachedBrokerLoadStats(boolean allowCapacityEstimation); ClusterModel clusterModel(ModelCompletenessRequirements requirements,
                                   boolean allowCapacityEstimation,
                                   OperationProgress operationProgress); ClusterModel clusterModel(long from,
                                   long to,
                                   ModelCompletenessRequirements requirements,
                                   boolean populateReplicaPlacementInfo,
                                   boolean allowCapacityEstimation,
                                   OperationProgress operationProgress); ClusterModel clusterCapacity(); void bootstrap(Long startMs, Long endMs, boolean clearMetrics); void pauseMetricSampling(String reason); void train(Long startMs, Long endMs); boolean setSelfHealingFor(AnomalyType anomalyType, boolean isSelfHealingEnabled); boolean setConcurrencyAdjusterFor(ConcurrencyType concurrencyType, boolean isConcurrencyAdjusterEnabled); boolean dropRecentBrokers(Set<Integer> brokersToDrop, boolean isRemoved); void addRecentBrokersPermanently(Set<Integer> brokersToAdd, boolean isRemoved); Set<Integer> recentBrokers(boolean isRemoved); void setRequestedExecutionProgressCheckIntervalMs(Long requestedExecutionProgressCheckIntervalMs); void setRequestedInterBrokerPartitionMovementConcurrency(Integer requestedInterBrokerPartitionMovementConcurrency); void setRequestedIntraBrokerPartitionMovementConcurrency(Integer requestedIntraBrokerPartitionMovementConcurrency); void setRequestedLeadershipMovementConcurrency(Integer requestedLeadershipMovementConcurrency); void resumeMetricSampling(String reason); OptimizerResult getProposals(OperationProgress operationProgress, boolean allowCapacityEstimation); boolean ignoreProposalCache(List<String> goals,
                                     ModelCompletenessRequirements requirements,
                                     Pattern excludedTopics,
                                     boolean excludeBrokers,
                                     boolean ignoreProposalCache,
                                     boolean isTriggeredByGoalViolation,
                                     Set<Integer> requestedDestinationBrokerIds,
                                     boolean isRebalanceDiskMode); synchronized OptimizerResult optimizations(ClusterModel clusterModel,
                                                    List<Goal> goalsByPriority,
                                                    OperationProgress operationProgress,
                                                    Map<TopicPartition, List<ReplicaPlacementInfo>> initReplicaDistribution,
                                                    OptimizationOptions optimizationOptions); Set<String> excludedTopics(ClusterModel clusterModel, Pattern requestedExcludedTopics); KafkaCruiseControlConfig config(); void executeProposals(Set<ExecutionProposal> proposals,
                               Set<Integer> unthrottledBrokers,
                               boolean isKafkaAssignerMode,
                               Integer concurrentInterBrokerPartitionMovements,
                               Integer concurrentIntraBrokerPartitionMovements,
                               Integer concurrentLeaderMovements,
                               Long executionProgressCheckIntervalMs,
                               ReplicaMovementStrategy replicaMovementStrategy,
                               Long replicationThrottle,
                               boolean isTriggeredByUserRequest,
                               String uuid,
                               boolean skipAutoRefreshingConcurrency); void executeRemoval(Set<ExecutionProposal> proposals,
                             boolean throttleDecommissionedBroker,
                             Set<Integer> removedBrokers,
                             boolean isKafkaAssignerMode,
                             Integer concurrentInterBrokerPartitionMovements,
                             Integer concurrentLeaderMovements,
                             Long executionProgressCheckIntervalMs,
                             ReplicaMovementStrategy replicaMovementStrategy,
                             Long replicationThrottle,
                             boolean isTriggeredByUserRequest,
                             String uuid); void executeDemotion(Set<ExecutionProposal> proposals,
                              Set<Integer> demotedBrokers,
                              Integer concurrentLeaderMovements,
                              int brokerCount,
                              Long executionProgressCheckIntervalMs,
                              ReplicaMovementStrategy replicaMovementStrategy,
                              Long replicationThrottle,
                              boolean isTriggeredByUserRequest,
                              String uuid); void userTriggeredStopExecution(boolean forceExecutionStop); void setGeneratingProposalsForExecution(String uuid, Supplier<String> reasonSupplier, boolean isTriggeredByUserRequest); synchronized void failGeneratingProposalsForExecution(String uuid); long executionProgressCheckIntervalMs(); ExecutorState.State executionState(); ExecutorState executorState(); LoadMonitorState monitorState(Cluster cluster); AnalyzerState analyzerState(Cluster cluster); AnomalyDetectorState anomalyDetectorState(); Cluster kafkaCluster(); TopicConfigProvider topicConfigProvider(); static String cruiseControlVersion(); static String cruiseControlCommitId(); ModelCompletenessRequirements modelCompletenessRequirements(Collection<Goal> goals); boolean meetCompletenessRequirements(List<String> goals); void sanityCheckBrokerPresence(Set<Integer> brokerIds);  }