public static void processElectLeadersResult(ElectLeadersResult result, Set<TopicPartition> deletedTopicPartitions) {
    if (result == null) {
      return;
    }
    try {
      Map<TopicPartition, Optional<Throwable>> partitions = result.partitions().get();
      Set<TopicPartition> noElectionNeeded = new HashSet<>();
      Set<TopicPartition> preferredLeaderUnavailable = new HashSet<>();

      for (Map.Entry<TopicPartition, Optional<Throwable>> entry : partitions.entrySet()) {
        TopicPartition tp = entry.getKey();
        if (entry.getValue().isEmpty()) {
          LOG.debug("Leader election for {} has succeeded.", tp);
        } else {
          if (Errors.ELECTION_NOT_NEEDED.exception().getClass() == entry.getValue().get().getClass()) {
            // The leader is already the preferred leader.
            noElectionNeeded.add(tp);
          } else if (Errors.UNKNOWN_TOPIC_OR_PARTITION.exception().getClass() == entry.getValue().get().getClass()
                     || Errors.INVALID_TOPIC_EXCEPTION.exception().getClass() == entry.getValue().get().getClass()) {
            // Topic (1) has been deleted -- i.e. since partition does not exist, it is assumed to be deleted or (2) is being deleted.
            deletedTopicPartitions.add(tp);
          } else if (Errors.PREFERRED_LEADER_NOT_AVAILABLE.exception().getClass() == entry.getValue().get().getClass()) {
            // We tried to elect the preferred leader but it is offline.
            preferredLeaderUnavailable.add(tp);
          } else {
            // Based on the KafkaAdminClient code, looks like there is no handling / retry in response to a Errors.NOT_CONTROLLER -- e.g. if
            // the controller changes during a request, the leader election will be dropped with an error response. In such cases, a
            // followup execution would be needed (i.e. see Executor#maybeReexecuteLeadershipTasks(Set<TopicPartition>)).
            LOG.warn("Failed to elect preferred leader for {}.", tp, entry.getValue().get());
          }
        }
      }

      // Log relevant information for debugging purposes.
      if (!noElectionNeeded.isEmpty()) {
        LOG.debug("Leader election not needed for {}.", noElectionNeeded);
      }
      if (!preferredLeaderUnavailable.isEmpty()) {
        LOG.debug("The preferred leader was not available for {}.", preferredLeaderUnavailable);
      }
      if (!deletedTopicPartitions.isEmpty()) {
        LOG.debug("Corresponding topics have been deleted before leader election {}.", deletedTopicPartitions);
      }
    } catch (ExecutionException ee) {
      if (Errors.REQUEST_TIMED_OUT.exception().getClass() == ee.getCause().getClass()) {
        throw new IllegalStateException(String.format("electLeaders request timed out. Check for Kafka broker- or controller-side issues and"
                                                      + " consider increasing the configured timeout (see %s).",
                                                      ExecutorConfig.ADMIN_CLIENT_REQUEST_TIMEOUT_MS_CONFIG), ee.getCause());
      } else if (Errors.CLUSTER_AUTHORIZATION_FAILED.exception().getClass() == ee.getCause().getClass()) {
        throw new IllegalStateException("Cruise Control is not authorized to trigger leader election", ee.getCause());
      } else {
        // Not expected to happen.
        throw new IllegalStateException("An unknown execution exception is encountered in response to electLeaders.", ee.getCause());
      }
    } catch (InterruptedException e) {
      LOG.warn("Interrupted during the process of ElectLeadersResult.", e);
    }
  }
@Test
  public void testProcessElectLeadersResult() throws Exception {
    // Case 1: Handle null result input. Expect no side effect.
    ExecutionUtils.processElectLeadersResult(null, Collections.emptySet());

    KafkaFutureImpl<Map<TopicPartition, Optional<Throwable>>> partitions = EasyMock.mock(KafkaFutureImpl.class);
    Constructor<ElectLeadersResult> constructor = null;
    try {
      constructor = ElectLeadersResult.class.getDeclaredConstructor(KafkaFuture.class);
    } catch (NoSuchMethodException e) {
      LOG.debug("Unable to find Kafka 3.0+ constructor for ElectLeaderResult class", e);
    }
    if (constructor == null) {
      try {
        constructor = ElectLeadersResult.class.getDeclaredConstructor(KafkaFutureImpl.class);
      } catch (NoSuchMethodException e) {
        LOG.debug("Unable to find Kafka 3.0- constructor for ElectLeaderResult class", e);
      }
    }
    if (constructor == null) {
      throw new NoSuchElementException("Unable to find viable constructor for the ElectionLeadersResult class");
    }
    constructor.setAccessible(true);
    ElectLeadersResult result;

    // Case 2: Handle both partitions are successful
    // Case 3: Handle one partition successful, the other has ElectionNotNeededException
    // Case 4: Handle one partition successful, the other has PreferredLeaderNotAvailableException
    // Case 5: Handle one partition successful, the other has NotControllerException
    for (Map<TopicPartition, Optional<Throwable>> entry : Set.of(SUCCESSFUL_PARTITIONS, ONE_WITH_ENN, ONE_WITH_PLNA, ONE_WITH_NC)) {
      result = constructor.newInstance(partitions);

      EasyMock.expect(partitions.get()).andReturn(entry).once();
      EasyMock.replay(partitions);

      ExecutionUtils.processElectLeadersResult(result, Collections.emptySet());
      EasyMock.verify(partitions);
      EasyMock.reset(partitions);
    }

    // Case 6: Handle one partition successful, the other has UnknownTopicOrPartitionException
    // Case 7: Handle one partition successful, the other has InvalidTopicException
    for (Map<TopicPartition, Optional<Throwable>> entry : Set.of(ONE_WITH_UTOP, ONE_WITH_IT)) {
      result = constructor.newInstance(partitions);

      EasyMock.expect(partitions.get()).andReturn(entry).once();
      EasyMock.replay(partitions);

      Set<TopicPartition> deletedTopicPartitions = new HashSet<>();
      ExecutionUtils.processElectLeadersResult(result, deletedTopicPartitions);
      Assert.assertEquals(1, deletedTopicPartitions.size());
      Assert.assertEquals(P1, deletedTopicPartitions.iterator().next());
      EasyMock.verify(partitions);
      EasyMock.reset(partitions);
    }

    // Case 8: Handle execution timeout exception. Expect no side effect.
    // Case 9: Handle execution ClusterAuthorization exception. Expect no side effect.
    // Case 10: Handle unexpected execution exception (i.e. ControllerMovedException). Expect no side effect.
    for (Throwable entry : Set.of(new org.apache.kafka.common.errors.TimeoutException(),
                                  new org.apache.kafka.common.errors.ClusterAuthorizationException(""),
                                  new org.apache.kafka.common.errors.ControllerMovedException(""))) {
      result = constructor.newInstance(partitions);

      EasyMock.expect(partitions.get()).andThrow(new ExecutionException(entry)).once();
      EasyMock.replay(partitions);

      ElectLeadersResult exceptionResult = result;
      Exception thrownException = assertThrows(IllegalStateException.class,
                                               () -> ExecutionUtils.processElectLeadersResult(exceptionResult, Collections.emptySet()));
      Assert.assertEquals(entry, thrownException.getCause());
      EasyMock.verify(partitions);
      EasyMock.reset(partitions);
    }

    // Case 11: Handle future wait interrupted exception
    result = constructor.newInstance(partitions);

    EasyMock.expect(partitions.get()).andThrow(new InterruptedException()).once();
    EasyMock.replay(partitions);

    ExecutionUtils.processElectLeadersResult(result, Collections.emptySet());
    EasyMock.verify(partitions);
    EasyMock.reset(partitions);
  }