ahuang98 commented on code in PR #22505:
URL: https://github.com/apache/kafka/pull/22505#discussion_r3389723267


##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2820,29 +2826,36 @@ private void handleRequest(RaftRequest.Inbound request, 
long currentTimeMs) {
             default -> throw new IllegalArgumentException("Unexpected request 
type " + apiKey);
         };
 
-        responseFuture.whenComplete((response, exception) -> {
-            ApiMessage message = response;
-            if (message == null) {
-                message = RaftUtil.errorResponse(apiKey, 
Errors.forException(exception));
-            }
+        responseStage.whenComplete((response, exception) -> {
+            var message = response == null ?
+                RaftUtil.errorResponse(apiKey, Errors.forException(exception)) 
:
+                response;
+
+            var raftResponse = new 
RaftResponse.Outbound(request.correlationId(), message);
 
-            RaftResponse.Outbound responseMessage = new 
RaftResponse.Outbound(request.correlationId(), message);
-            request.completion.complete(responseMessage);
-            logger.trace("Sent response {} to inbound request {}", 
responseMessage, request);
+            future.complete(raftResponse);
+            logger.trace("Sent response {} to inbound request {}", 
raftResponse, request);
         });
     }
 
-    private void handleInboundMessage(RaftMessage message, long currentTimeMs) 
{
+    private void handleInboundMessage(
+        RaftMessage message,
+        CompletableFuture<RaftMessage> future,
+        long currentTimeMs
+    ) {
         logger.trace("Received inbound message {}", message);
 
         if (message instanceof RaftRequest.Inbound request) {
-            handleRequest(request, currentTimeMs);
+            handleRequest(request, future, currentTimeMs);
         } else if (message instanceof RaftResponse.Inbound response) {
             if (requestManager.isResponseExpected(response.source(), 
response.correlationId())) {
                 handleResponse(response, currentTimeMs);
             } else {
                 logger.debug("Ignoring response {} since it is no longer 
needed", response);
             }
+            // Inboud response messages do not have an associated outbound 
message. Nothing should

Review Comment:
   sounds good



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to