public synchronized void shutdown() {
    LOG.info("Shutting down executor.");

    try {
      _flipOngoingExecutionMutex.acquire();
      try {
        if (_hasOngoingExecution) {
          LOG.warn("Shutdown executor may take long because execution is still in progress.");
          stopExecution();
        }
        try {
          LOG.info("Waiting for ongoing execution to stop completely if any.");
          _noOngoingExecutionSemaphore.acquire();
          LOG.info("Blocking new execution from proceeding.");
        } catch (final InterruptedException ie) {
          LOG.warn("Interrupted while waiting for ongoing execution to stop.");
        }
      } finally {
        _flipOngoingExecutionMutex.release();
      }
    } catch (final InterruptedException ie) {
      // handle acquire failure for _flipOngoingExecutionMutex
      LOG.warn("Interrupted while checking if there is ongoing execution to stop.");
    }

    _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.closeAdminClientWithTimeout(_adminClient);
    _executionHistoryScannerExecutor.shutdownNow();
    _concurrencyAdjusterExecutor.shutdownNow();
    _topicMinIsrCache.shutdown();
    LOG.info("Executor shutdown completed.");
  }
@Test
  public void testSubmitPreferredLeaderElection() throws InterruptedException, ExecutionException {
    Map<String, TopicDescription> topicDescriptions = createTopics(0);
    AdminClient adminClient = KafkaCruiseControlUtils.createAdminClient(Collections.singletonMap(
        AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, broker(0).plaintextAddr()));

    // Case 1: Handle the successful case.
    ReplicaPlacementInfo oldLeader = new ReplicaPlacementInfo(topicDescriptions.get(TP1.topic()).partitions().get(PARTITION).leader().id());
    List<ReplicaPlacementInfo> replicas = new ArrayList<>();
    for (Node node : topicDescriptions.get(TP1.topic()).partitions().get(PARTITION).replicas()) {
      replicas.add(new ReplicaPlacementInfo(node.id()));
    }

    List<ReplicaPlacementInfo> newReplicas = new ArrayList<>(replicas.size());
    for (int i = replicas.size() - 1; i >= 0; i--) {
      newReplicas.add(new ReplicaPlacementInfo(replicas.get(i).brokerId()));
    }

    // Reorder replicas before leadership election to change the preferred replica.
    ExecutionTask swapTask = new ExecutionTask(0,
                                           new ExecutionProposal(TP1, 0, oldLeader, replicas, newReplicas),
                                           ExecutionTask.TaskType.INTER_BROKER_REPLICA_ACTION,
                                           EXECUTION_ALERTING_THRESHOLD_MS);
    swapTask.inProgress(MOCK_CURRENT_TIME);

    AlterPartitionReassignmentsResult swapResult = ExecutionUtils.submitReplicaReassignmentTasks(adminClient, Collections.singletonList(swapTask));

    assertEquals(1, swapResult.values().size());
    // Can retrieve the future if it is successful.
    swapResult.values().get(TP1).get();

    // Leadership task to transfer the leadership to the preferred replica.
    ExecutionTask task = new ExecutionTask(0,
                                           new ExecutionProposal(TP1, 0, oldLeader, newReplicas, newReplicas),
                                           ExecutionTask.TaskType.LEADER_ACTION,
                                           EXECUTION_ALERTING_THRESHOLD_MS);
    task.inProgress(MOCK_CURRENT_TIME);

    ElectLeadersResult result = ExecutionUtils.submitPreferredLeaderElection(adminClient, Collections.singletonList(task));
    assertTrue(result.partitions().get().get(TP1).isEmpty());

    // Case 2: Handle "election not needed" -- i.e. the leader is already the preferred replica for TP0.
    oldLeader = new ReplicaPlacementInfo(topicDescriptions.get(TP0.topic()).partitions().get(PARTITION).leader().id());
    replicas.clear();
    for (Node node : topicDescriptions.get(TP0.topic()).partitions().get(PARTITION).replicas()) {
      replicas.add(new ReplicaPlacementInfo(node.id()));
    }
    task = new ExecutionTask(0,
                             new ExecutionProposal(TP0, 0, oldLeader, replicas, replicas),
                             ExecutionTask.TaskType.LEADER_ACTION,
                             EXECUTION_ALERTING_THRESHOLD_MS);
    task.inProgress(MOCK_CURRENT_TIME);

    result = ExecutionUtils.submitPreferredLeaderElection(adminClient, Collections.singletonList(task));
    assertTrue(result.partitions().get().get(TP0).isPresent());
    assertEquals(Errors.ELECTION_NOT_NEEDED.exception().getClass(), result.partitions().get().get(TP0).get().getClass());

    // Case 3: Handle "unknown topic or partition" -- i.e. the partition TP2 does not exist, maybe it existed at some point and then deleted
    task = new ExecutionTask(0,
                             new ExecutionProposal(TP2, 0, oldLeader, replicas, replicas),
                             ExecutionTask.TaskType.LEADER_ACTION,
                             EXECUTION_ALERTING_THRESHOLD_MS);
    task.inProgress(MOCK_CURRENT_TIME);

    result = ExecutionUtils.submitPreferredLeaderElection(adminClient, Collections.singletonList(task));
    assertTrue(result.partitions().get().get(TP2).isPresent());
    assertEquals(Errors.UNKNOWN_TOPIC_OR_PARTITION.exception().getClass(), result.partitions().get().get(TP2).get().getClass());

    // Case 4: Handle the scenario, where we tried to elect the preferred leader but it is offline.
    _brokers.get(0).shutdown();
    task = new ExecutionTask(0,
                             new ExecutionProposal(TP0, 0, oldLeader, replicas, replicas),
                             ExecutionTask.TaskType.LEADER_ACTION,
                             EXECUTION_ALERTING_THRESHOLD_MS);

    task.inProgress(MOCK_CURRENT_TIME);

    result = ExecutionUtils.submitPreferredLeaderElection(adminClient, Collections.singletonList(task));
    assertTrue(result.partitions().get().get(TP0).isPresent());
    assertEquals(Errors.PREFERRED_LEADER_NOT_AVAILABLE.exception().getClass(), result.partitions().get().get(TP0).get().getClass());
  }