KAFKA-19203 Replace ApiError#exception by Error#exception for KafkaAdminClient (#19623)

This pull request addresses KAFKA-19203 by replacing
`ApiError#exception` with `Error#exception` in `KafkaAdminClient`. The
previous use of `ApiError#exception` was redundant, as we only need the
exception without the additional wrapping of `ApiError`.

## Changes

- Replaced some usages of `ApiError#exception` with `Error#exception` in
`KafkaAdminClient`.
- Simplified exception handling logic to reduce unnecessary layers.

Reviewers: Chia-Ping Tsai <[email protected]>

Author: jim0987795064

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -2494,8 +2494,7 @@ void handleResponse(AbstractResponse abstractResponse) {
                     DescribeClusterResponse response = (DescribeClusterResponse) abstractResponse;
                     Errors error = Errors.forCode(response.data().errorCode());
                     if (error != Errors.NONE) {
-                        ApiError apiError = new ApiError(error, response.data().errorMessage());
-                        handleFailure(apiError.exception());
+                        handleFailure(error.exception(response.data().errorMessage()));
                         return;
                     }
 
@@ -2691,10 +2690,9 @@ void handleResponse(AbstractResponse abstractResponse) {
                         } else {
                             List<FilterResult> filterResults = new ArrayList<>();
                             for (DeleteAclsMatchingAcl matchingAcl : filterResult.matchingAcls()) {
-                                ApiError aclError = new ApiError(Errors.forCode(matchingAcl.errorCode()),
-                                    matchingAcl.errorMessage());
+                                Errors aclError = Errors.forCode(matchingAcl.errorCode());
                                 AclBinding aclBinding = DeleteAclsResponse.aclBinding(matchingAcl);
-                                filterResults.add(new FilterResult(aclBinding, aclError.exception()));
+                                filterResults.add(new FilterResult(aclBinding, aclError.exception(matchingAcl.errorMessage())));
                             }
                             future.complete(new FilterResults(filterResults));
                         }
@@ -3995,7 +3993,7 @@ public void handleResponse(AbstractResponse abstractResponse) {
                             for (ReassignablePartitionResponse partition : topicResponse.partitions()) {
                                 errors.put(
                                     new TopicPartition(topicName, partition.partitionIndex()),
-                                    new ApiError(topLevelError, response.data().errorMessage()).exception()
+                                    topLevelError.exception(response.data().errorMessage())
                                 );
                                 receivedResponsesCount += 1;
                             }
@@ -4035,7 +4033,7 @@ private int validateTopicResponses(List<ReassignableTopicResponse> topicResponse
                         if (partitionError == Errors.NONE) {
                             errors.put(tp, null);
                         } else {
-                            errors.put(tp, new ApiError(partitionError, partResponse.errorMessage()).exception());
+                            errors.put(tp, partitionError.exception(partResponse.errorMessage()));
                         }
                         receivedResponsesCount += 1;
                     }
@@ -4111,7 +4109,7 @@ void handleResponse(AbstractResponse abstractResponse) {
                         handleNotControllerError(error);
                         break;
                     default:
-                        partitionReassignmentsFuture.completeExceptionally(new ApiError(error, response.data().errorMessage()).exception());
+                        partitionReassignmentsFuture.completeExceptionally(error.exception(response.data().errorMessage()));
                         break;
                 }
                 Map<TopicPartition, PartitionReassignment> reassignmentMap = new HashMap<>();
@@ -4993,14 +4991,11 @@ AddRaftVoterRequest.Builder createRequest(int timeoutMs) {
             void handleResponse(AbstractResponse response) {
                 handleNotControllerError(response);
                 AddRaftVoterResponse addResponse = (AddRaftVoterResponse) response;
-                if (addResponse.data().errorCode() != Errors.NONE.code()) {
-                    ApiError error = new ApiError(
-                        addResponse.data().errorCode(),
-                        addResponse.data().errorMessage());
-                    future.completeExceptionally(error.exception());
-                } else {
+                Errors error = Errors.forCode(addResponse.data().errorCode());
+                if (error != Errors.NONE)
+                    future.completeExceptionally(error.exception(addResponse.data().errorMessage()));
+                else
                     future.complete(null);
-                }
             }
 
             @Override
@@ -5038,14 +5033,11 @@ RemoveRaftVoterRequest.Builder createRequest(int timeoutMs) {
             void handleResponse(AbstractResponse response) {
                 handleNotControllerError(response);
                 RemoveRaftVoterResponse addResponse = (RemoveRaftVoterResponse) response;
-                if (addResponse.data().errorCode() != Errors.NONE.code()) {
-                    ApiError error = new ApiError(
-                        addResponse.data().errorCode(),
-                        addResponse.data().errorMessage());
-                    future.completeExceptionally(error.exception());
-                } else {
+                Errors error = Errors.forCode(addResponse.data().errorCode());
+                if (error != Errors.NONE)
+                    future.completeExceptionally(error.exception(addResponse.data().errorMessage()));
+                else
                     future.complete(null);
-                }
             }
 
             @Override