public static void processAlterPartitionReassignmentsResult(AlterPartitionReassignmentsResult result,
                                                              Set<TopicPartition> deletedTopicPartitions,
                                                              Set<TopicPartition> deadTopicPartitions,
                                                              Set<TopicPartition> noReassignmentToCancelTopicPartitions) {
    if (result == null) {
      return;
    }
    for (Map.Entry<TopicPartition, KafkaFuture<Void>> entry: result.values().entrySet()) {
      TopicPartition tp = entry.getKey();
      try {
        entry.getValue().get();
        LOG.debug("Replica reassignment for {} has been accepted.", tp);
      } catch (ExecutionException ee) {
        if (Errors.INVALID_REPLICA_ASSIGNMENT.exception().getClass() == ee.getCause().getClass()) {
          deadTopicPartitions.add(tp);
          LOG.debug("Replica reassignment failed for {} due to dead destination broker(s).", tp);
        } else if (Errors.UNKNOWN_TOPIC_OR_PARTITION.exception().getClass() == ee.getCause().getClass()) {
          deletedTopicPartitions.add(tp);
          LOG.debug("Replica reassignment failed for {} due to its unknown topic or partition.", tp);
        } else if (Errors.NO_REASSIGNMENT_IN_PROGRESS.exception().getClass() == ee.getCause().getClass()) {
          // Attempt to cancel/rollback a reassignment that does not exist.
          noReassignmentToCancelTopicPartitions.add(tp);
          LOG.debug("Rollback failed for {} due to lack of corresponding ongoing replica reassignment.", tp);
        } else if (Errors.REQUEST_TIMED_OUT.exception().getClass() == ee.getCause().getClass()) {
          throw new IllegalStateException(String.format("alterPartitionReassignments request timed out for partitions: %s. Check for Kafka "
                                                        + "broker- or controller-side issues and consider increasing the configured timeout "
                                                        + "(see %s).",
                                                        result.values().keySet(),
                                                        ExecutorConfig.ADMIN_CLIENT_REQUEST_TIMEOUT_MS_CONFIG), ee.getCause());
        } else {
          // Not expected to happen.
          throw new IllegalStateException(String.format("%s encountered an unknown execution exception.", tp), ee.getCause());
        }
      } catch (InterruptedException e) {
        LOG.warn("Interrupted during the process of AlterPartitionReassignmentsResult on {}.", result.values().keySet(), e);
      }
    }
  }
@Test
  public void testProcessAlterPartitionReassignmentResult() throws Exception {
    // Case 1: Handle null result input. Expect no side effect.
    ExecutionUtils.processAlterPartitionReassignmentsResult(null, Collections.emptySet(), Collections.emptySet(), Collections.emptySet());

    // Case 2: Handle successful execution results
    AlterPartitionReassignmentsResult result = EasyMock.mock(AlterPartitionReassignmentsResult.class);
    EasyMock.expect(result.values())
            .andReturn(getKafkaFutureByTopicPartition(null))
            .once();
    EasyMock.replay(result);
    ExecutionUtils.processAlterPartitionReassignmentsResult(result, Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
    EasyMock.verify(result);
    EasyMock.reset(result);

    // Case 3: Handle invalid replica assignment exception
    EasyMock.expect(result.values())
            .andReturn(getKafkaFutureByTopicPartition(new ExecutionException(Errors.INVALID_REPLICA_ASSIGNMENT.exception())))
            .once();
    EasyMock.replay(result);
    Set<TopicPartition> deadTopicPartitions = new HashSet<>();
    ExecutionUtils.processAlterPartitionReassignmentsResult(result, Collections.emptySet(), deadTopicPartitions, Collections.emptySet());
    assertEquals(Collections.singleton(new TopicPartition(TOPIC_NAME, P0.partition())), deadTopicPartitions);
    EasyMock.verify(result);
    EasyMock.reset(result);

    // Case 4: Handle unknown topic or partition exception
    EasyMock.expect(result.values())
            .andReturn(getKafkaFutureByTopicPartition(new ExecutionException(Errors.UNKNOWN_TOPIC_OR_PARTITION.exception())))
            .once();
    EasyMock.replay(result);
    Set<TopicPartition> deletedTopicPartitions = new HashSet<>();
    ExecutionUtils.processAlterPartitionReassignmentsResult(result, deletedTopicPartitions, Collections.emptySet(), Collections.emptySet());
    assertEquals(Collections.singleton(new TopicPartition(TOPIC_NAME, P0.partition())), deletedTopicPartitions);
    EasyMock.verify(result);
    EasyMock.reset(result);

    // Case 5: Handle no reassign in progress exception
    EasyMock.expect(result.values())
            .andReturn(getKafkaFutureByTopicPartition(new ExecutionException(Errors.NO_REASSIGNMENT_IN_PROGRESS.exception())))
            .once();
    EasyMock.replay(result);
    Set<TopicPartition> noReassignmentToCancelTopicPartitions = new HashSet<>();
    ExecutionUtils.processAlterPartitionReassignmentsResult(result,
                                                            Collections.emptySet(),
                                                            Collections.emptySet(),
                                                            noReassignmentToCancelTopicPartitions);
    assertEquals(Collections.singleton(new TopicPartition(TOPIC_NAME, P0.partition())), noReassignmentToCancelTopicPartitions);
    EasyMock.verify(result);
    EasyMock.reset(result);

    // Case 6: Handle execution timeout exception. Expect no side effect.
    org.apache.kafka.common.errors.TimeoutException kafkaTimeoutException = new org.apache.kafka.common.errors.TimeoutException();
    EasyMock.expect(result.values())
            .andReturn(getKafkaFutureByTopicPartition(new ExecutionException(kafkaTimeoutException)))
            .times(2);
    EasyMock.replay(result);
    Exception thrownException = assertThrows(IllegalStateException.class, () -> {
      ExecutionUtils.processAlterPartitionReassignmentsResult(result, Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
    });
    Assert.assertEquals(kafkaTimeoutException, thrownException.getCause());
    EasyMock.verify(result);
    EasyMock.reset(result);

    // Case 7: Handle future wait interrupted exception
    EasyMock.expect(result.values())
            .andReturn(getKafkaFutureByTopicPartition(new InterruptedException()))
            .times(2);
    EasyMock.replay(result);
    ExecutionUtils.processAlterPartitionReassignmentsResult(result, Collections.emptySet(), Collections.emptySet(), Collections.emptySet());
    EasyMock.verify(result);
    EasyMock.reset(result);
  }