@Test public void testBrokerDiesBeforeMovingPartition() throws Exception { KafkaZkClient kafkaZkClient = KafkaCruiseControlUtils.createKafkaZkClient(zookeeper().connectionString(), "ExecutorTestMetricGroup", "BrokerDiesBeforeMovingPartition", false); try { Map<String, TopicDescription> topicDescriptions = createTopics((int) PRODUCE_SIZE_IN_BYTES); int initialLeader0 = topicDescriptions.get(TOPIC0).partitions().get(0).leader().id(); int initialLeader1 = topicDescriptions.get(TOPIC1).partitions().get(0).leader().id(); _brokers.get(initialLeader0 == 0 ? 1 : 0).shutdown(); ExecutionProposal proposal0 = new ExecutionProposal(TP0, PRODUCE_SIZE_IN_BYTES, new ReplicaPlacementInfo(initialLeader0), Collections.singletonList(new ReplicaPlacementInfo(initialLeader0)), Collections.singletonList(new ReplicaPlacementInfo(initialLeader0 == 0 ? 1 : 0))); ExecutionProposal proposal1 = new ExecutionProposal(TP1, 0, new ReplicaPlacementInfo(initialLeader1), Arrays.asList(new ReplicaPlacementInfo(initialLeader1), new ReplicaPlacementInfo(initialLeader1 == 0 ? 1 : 0)), Arrays.asList(new ReplicaPlacementInfo(initialLeader1 == 0 ? 1 : 0), new ReplicaPlacementInfo(initialLeader1))); Collection<ExecutionProposal> proposalsToExecute = Arrays.asList(proposal0, proposal1); executeAndVerifyProposals(kafkaZkClient, proposalsToExecute, Collections.emptyList(), true, null, false, false); assertEquals(Collections.singletonList(initialLeader0 == 0 ? 1 : 0), ExecutorUtils.newAssignmentForPartition(kafkaZkClient, TP0)); assertEquals(initialLeader0, kafkaZkClient.getLeaderForPartition(TP0).get()); assertEquals(initialLeader0, kafkaZkClient.getLeaderForPartition(TP1).get()); } finally { KafkaCruiseControlUtils.closeKafkaZkClientWithTimeout(kafkaZkClient); } }
public synchronized void shutdown() { LOG.info("Shutting down executor."); if (_hasOngoingExecution) { LOG.warn("Shutdown executor may take long because execution is still in progress."); } _proposalExecutor.shutdown(); try { _proposalExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _metadataClient.close(); KafkaCruiseControlUtils.closeKafkaZkClientWithTimeout(_kafkaZkClient); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); _executionHistoryScannerExecutor.shutdownNow(); _concurrencyAdjusterExecutor.shutdownNow(); LOG.info("Executor shutdown completed."); }
Executor { public synchronized void shutdown() { LOG.info("Shutting down executor."); if (_hasOngoingExecution) { LOG.warn("Shutdown executor may take long because execution is still in progress."); } _proposalExecutor.shutdown(); try { _proposalExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _metadataClient.close(); KafkaCruiseControlUtils.closeKafkaZkClientWithTimeout(_kafkaZkClient); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); _executionHistoryScannerExecutor.shutdownNow(); _concurrencyAdjusterExecutor.shutdownNow(); LOG.info("Executor shutdown completed."); } }
Executor { public synchronized void shutdown() { LOG.info("Shutting down executor."); if (_hasOngoingExecution) { LOG.warn("Shutdown executor may take long because execution is still in progress."); } _proposalExecutor.shutdown(); try { _proposalExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _metadataClient.close(); KafkaCruiseControlUtils.closeKafkaZkClientWithTimeout(_kafkaZkClient); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); _executionHistoryScannerExecutor.shutdownNow(); _concurrencyAdjusterExecutor.shutdownNow(); LOG.info("Executor shutdown completed."); } Executor(KafkaCruiseControlConfig config,
                  Time time,
                  MetricRegistry dropwizardMetricRegistry,
                  AnomalyDetectorManager anomalyDetectorManager);   Executor(KafkaCruiseControlConfig config,
           Time time,
           MetricRegistry dropwizardMetricRegistry,
           MetadataClient metadataClient,
           ExecutorNotifier executorNotifier,
           AnomalyDetectorManager anomalyDetectorManager); }
Executor { public synchronized void shutdown() { LOG.info("Shutting down executor."); if (_hasOngoingExecution) { LOG.warn("Shutdown executor may take long because execution is still in progress."); } _proposalExecutor.shutdown(); try { _proposalExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _metadataClient.close(); KafkaCruiseControlUtils.closeKafkaZkClientWithTimeout(_kafkaZkClient); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); _executionHistoryScannerExecutor.shutdownNow(); _concurrencyAdjusterExecutor.shutdownNow(); LOG.info("Executor shutdown completed."); } Executor(KafkaCruiseControlConfig config,
                  Time time,
                  MetricRegistry dropwizardMetricRegistry,
                  AnomalyDetectorManager anomalyDetectorManager);   Executor(KafkaCruiseControlConfig config,
           Time time,
           MetricRegistry dropwizardMetricRegistry,
           MetadataClient metadataClient,
           ExecutorNotifier executorNotifier,
           AnomalyDetectorManager anomalyDetectorManager); Set<ExecutionTask> inExecutionTasks(); synchronized void setRequestedExecutionProgressCheckIntervalMs(Long requestedExecutionProgressCheckIntervalMs); long executionProgressCheckIntervalMs(); Set<Integer> recentlyDemotedBrokers(); Set<Integer> recentlyRemovedBrokers(); boolean dropRecentlyRemovedBrokers(Set<Integer> brokersToDrop); boolean dropRecentlyDemotedBrokers(Set<Integer> brokersToDrop); void addRecentlyRemovedBrokers(Set<Integer> brokersToAdd); void addRecentlyDemotedBrokers(Set<Integer> brokersToAdd); ExecutorState state(); boolean setConcurrencyAdjusterFor(ConcurrencyType concurrencyType, boolean isConcurrencyAdjusterEnabled); synchronized void executeProposals(Collection<ExecutionProposal> proposals,
                                            Set<Integer> unthrottledBrokers,
                                            Set<Integer> removedBrokers,
                                            LoadMonitor loadMonitor,
                                            Integer requestedInterBrokerPartitionMovementConcurrency,
                                            Integer requestedIntraBrokerPartitionMovementConcurrency,
                                            Integer requestedLeadershipMovementConcurrency,
                                            Long requestedExecutionProgressCheckIntervalMs,
                                            ReplicaMovementStrategy replicaMovementStrategy,
                                            Long replicationThrottle,
                                            boolean isTriggeredByUserRequest,
                                            String uuid,
                                            boolean isKafkaAssignerMode,
                                            boolean skipAutoRefreshingConcurrency); synchronized void executeDemoteProposals(Collection<ExecutionProposal> proposals,
                                                  Collection<Integer> demotedBrokers,
                                                  LoadMonitor loadMonitor,
                                                  Integer concurrentSwaps,
                                                  Integer requestedLeadershipMovementConcurrency,
                                                  Long requestedExecutionProgressCheckIntervalMs,
                                                  ReplicaMovementStrategy replicaMovementStrategy,
                                                  Long replicationThrottle,
                                                  boolean isTriggeredByUserRequest,
                                                  String uuid); void setRequestedInterBrokerPartitionMovementConcurrency(Integer requestedInterBrokerPartitionMovementConcurrency); void setRequestedIntraBrokerPartitionMovementConcurrency(Integer requestedIntraBrokerPartitionMovementConcurrency); void setRequestedLeadershipMovementConcurrency(Integer requestedLeadershipMovementConcurrency); void setUserTaskManager(UserTaskManager userTaskManager); synchronized void setGeneratingProposalsForExecution(String uuid, Supplier<String> reasonSupplier, boolean isTriggeredByUserRequest); synchronized void failGeneratingProposalsForExecution(String uuid); synchronized void userTriggeredStopExecution(boolean forceExecutionStop); synchronized void shutdown(); boolean modifyOngoingExecution(boolean modify); boolean hasOngoingExecution(); boolean hasOngoingPartitionReassignments(); boolean hasOngoingLeaderElection(); }
Executor { public synchronized void shutdown() { LOG.info("Shutting down executor."); if (_hasOngoingExecution) { LOG.warn("Shutdown executor may take long because execution is still in progress."); } _proposalExecutor.shutdown(); try { _proposalExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { LOG.warn("Interrupted while waiting for anomaly detector to shutdown."); } _metadataClient.close(); KafkaCruiseControlUtils.closeKafkaZkClientWithTimeout(_kafkaZkClient); KafkaCruiseControlUtils.closeAdminClientWithTimeout(_adminClient); _executionHistoryScannerExecutor.shutdownNow(); _concurrencyAdjusterExecutor.shutdownNow(); LOG.info("Executor shutdown completed."); } Executor(KafkaCruiseControlConfig config,
                  Time time,
                  MetricRegistry dropwizardMetricRegistry,
                  AnomalyDetectorManager anomalyDetectorManager);   Executor(KafkaCruiseControlConfig config,
           Time time,
           MetricRegistry dropwizardMetricRegistry,
           MetadataClient metadataClient,
           ExecutorNotifier executorNotifier,
           AnomalyDetectorManager anomalyDetectorManager); Set<ExecutionTask> inExecutionTasks(); synchronized void setRequestedExecutionProgressCheckIntervalMs(Long requestedExecutionProgressCheckIntervalMs); long executionProgressCheckIntervalMs(); Set<Integer> recentlyDemotedBrokers(); Set<Integer> recentlyRemovedBrokers(); boolean dropRecentlyRemovedBrokers(Set<Integer> brokersToDrop); boolean dropRecentlyDemotedBrokers(Set<Integer> brokersToDrop); void addRecentlyRemovedBrokers(Set<Integer> brokersToAdd); void addRecentlyDemotedBrokers(Set<Integer> brokersToAdd); ExecutorState state(); boolean setConcurrencyAdjusterFor(ConcurrencyType concurrencyType, boolean isConcurrencyAdjusterEnabled); synchronized void executeProposals(Collection<ExecutionProposal> proposals,
                                            Set<Integer> unthrottledBrokers,
                                            Set<Integer> removedBrokers,
                                            LoadMonitor loadMonitor,
                                            Integer requestedInterBrokerPartitionMovementConcurrency,
                                            Integer requestedIntraBrokerPartitionMovementConcurrency,
                                            Integer requestedLeadershipMovementConcurrency,
                                            Long requestedExecutionProgressCheckIntervalMs,
                                            ReplicaMovementStrategy replicaMovementStrategy,
                                            Long replicationThrottle,
                                            boolean isTriggeredByUserRequest,
                                            String uuid,
                                            boolean isKafkaAssignerMode,
                                            boolean skipAutoRefreshingConcurrency); synchronized void executeDemoteProposals(Collection<ExecutionProposal> proposals,
                                                  Collection<Integer> demotedBrokers,
                                                  LoadMonitor loadMonitor,
                                                  Integer concurrentSwaps,
                                                  Integer requestedLeadershipMovementConcurrency,
                                                  Long requestedExecutionProgressCheckIntervalMs,
                                                  ReplicaMovementStrategy replicaMovementStrategy,
                                                  Long replicationThrottle,
                                                  boolean isTriggeredByUserRequest,
                                                  String uuid); void setRequestedInterBrokerPartitionMovementConcurrency(Integer requestedInterBrokerPartitionMovementConcurrency); void setRequestedIntraBrokerPartitionMovementConcurrency(Integer requestedIntraBrokerPartitionMovementConcurrency); void setRequestedLeadershipMovementConcurrency(Integer requestedLeadershipMovementConcurrency); void setUserTaskManager(UserTaskManager userTaskManager); synchronized void setGeneratingProposalsForExecution(String uuid, Supplier<String> reasonSupplier, boolean isTriggeredByUserRequest); synchronized void failGeneratingProposalsForExecution(String uuid); synchronized void userTriggeredStopExecution(boolean forceExecutionStop); synchronized void shutdown(); boolean modifyOngoingExecution(boolean modify); boolean hasOngoingExecution(); boolean hasOngoingPartitionReassignments(); boolean hasOngoingLeaderElection(); final long _slowTaskAlertingBackoffTimeMs; }