ahuang98 commented on code in PR #22505:
URL: https://github.com/apache/kafka/pull/22505#discussion_r3378238636
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1542,12 +1546,10 @@ private CompletableFuture<FetchResponseData>
handleFetchRequest(
return completedFuture(response);
}
- CompletableFuture<Long> future = fetchPurgatory.await(
+ return fetchPurgatory.await(
Review Comment:
nit, seems like FuturePurgatory#await should also return `CompletionStage`
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -91,6 +92,11 @@ public int epoch() {
return epoch;
}
+ @Override
Review Comment:
should we mark FollowerState final too?
##########
raft/src/main/java/org/apache/kafka/raft/internals/BlockingMessageQueue.java:
##########
@@ -17,60 +17,51 @@
package org.apache.kafka.raft.internals;
import org.apache.kafka.common.errors.InterruptException;
-import org.apache.kafka.common.protocol.ApiMessage;
-import org.apache.kafka.raft.RaftMessage;
import org.apache.kafka.raft.RaftMessageQueue;
+import java.util.Optional;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class BlockingMessageQueue implements RaftMessageQueue {
- private static final RaftMessage WAKEUP_MESSAGE = new RaftMessage() {
- @Override
- public int correlationId() {
- return 0;
- }
-
- @Override
- public ApiMessage data() {
- return null;
- }
- };
-
- private final BlockingQueue<RaftMessage> queue = new
LinkedBlockingQueue<>();
- private final AtomicInteger size = new AtomicInteger(0);
+ private final BlockingQueue<QueueEntry> queue = new
LinkedBlockingQueue<>();
+ private final AtomicInteger messageCount = new AtomicInteger(0);
@Override
- public RaftMessage poll(long timeoutMs) {
+ public Optional<QueueEntry> poll(long timeoutMs) {
try {
- RaftMessage message = queue.poll(timeoutMs, TimeUnit.MILLISECONDS);
- if (message == null || message == WAKEUP_MESSAGE) {
- return null;
- } else {
- size.decrementAndGet();
- return message;
+ var entry = queue.poll(timeoutMs, TimeUnit.MILLISECONDS);
+ while (entry != null && entry.message() == null) {
+ // Drain the queue of all of the wakeup events
+ entry = queue.poll();
+ }
+ if (entry != null) {
+ messageCount.decrementAndGet();
}
+ return Optional.ofNullable(entry);
} catch (InterruptedException e) {
throw new InterruptException(e);
}
}
@Override
- public void add(RaftMessage message) {
- queue.add(message);
- size.incrementAndGet();
+ public void add(QueueEntry entry) {
+ queue.add(entry);
+ if (entry.message() != null) {
+ messageCount.incrementAndGet();
+ }
}
@Override
public boolean isEmpty() {
- return size.get() == 0;
+ return messageCount.get() == 0;
}
@Override
public void wakeup() {
- queue.add(WAKEUP_MESSAGE);
+ queue.add(new QueueEntry(null));
Review Comment:
what about changing QueueEntry into a sealed interface that permits either
`MessageEntry` or `WakeupEntry`? currently the wakeup event is a bit implicit
(`null` message means wakeup) and we create a throwaway completable future for
the wakeup event
##########
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:
interesting, maybe then
```
sealed interface QueueElement permits InboundRequestEntry,
InboundResponseEntry, Wakeup
```
?
--
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]
