jsancio commented on code in PR #22505:
URL: https://github.com/apache/kafka/pull/22505#discussion_r3375333918
##########
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();
+ }
Review Comment:
I was originally just fixing the API but decided to change the semantic of
this call for two reasons:
1. The fact that a message is used to implement wake up is an implementation
detail. The user should not be able to observer this. They can observe this
when isEmpty is false yet poll would return null in the old implementation.
2. The old implementation would cause the KRaft state machine
(KafkaRaftClient::pollCurrentState) to get polled once per wake up call even
though there are no new message or the timers have not expired. Conceptually,
multiple wake ups because two calls to poll can be collated into one wake up to
poll.
I am trying to make it more of a level trigger vs an edge trigger.
##########
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();
+ }
Review Comment:
The new code treats any null message as a wake up entry. The other solution
is to assert that `entry.message()` is not null. Let me fix this to make it
clearer.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2883,24 +2896,12 @@ private RequestSendResult maybeSendRequest(
currentTimeMs
);
- requestMessage.completion.whenComplete((response, exception) -> {
- if (exception != null) {
- ApiKeys api = ApiKeys.forId(request.apiKey());
- Errors error = Errors.forException(exception);
- ApiMessage errorResponse = RaftUtil.errorResponse(api,
error);
-
- response = new RaftResponse.Inbound(
- correlationId,
- errorResponse,
- destination
- );
Review Comment:
Correct. I took a look at the implementation and given the contract for
RequestAndCompletionHandler, KafkaNetworkChannel always gets a response. That
means that the associated future always completes with a value and never with
an exception.
I updated the documentation for NetworkChannel to document this.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -3677,14 +3690,12 @@ public void poll() {
long startWaitTimeMs = time.milliseconds();
kafkaRaftMetrics.updatePollStart(startWaitTimeMs);
- RaftMessage message = messageQueue.poll(pollTimeoutMs);
+ var maybeEntry = messageQueue.poll(pollTimeoutMs);
Review Comment:
MessageQueue (BlockingMessageQueue) should never return (new
QueueEntry(null). The implementation keeps polling until message is not null or
the queue is empty.
--
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]
