jsancio commented on code in PR #22505:
URL: https://github.com/apache/kafka/pull/22505#discussion_r3388712525
##########
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:
That works. Can we do this in a future PR? Ultimately, I would like to
replace MessageQueue with an generic event loop where you can do
`CompletionStage<T> submit(Callable<T> task);`
Inbound requests would be `Callable<RaftResponse.Outbound>`. Inbound
responses would be `Callable<Void>` what do you think?
--
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]
