jsancio commented on code in PR #22505:
URL: https://github.com/apache/kafka/pull/22505#discussion_r3430931891
##########
raft/src/main/java/org/apache/kafka/raft/internals/BlockingMessageQueue.java:
##########
@@ -17,60 +17,104 @@
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.CompletableFuture;
+import java.util.concurrent.CompletionStage;
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() {
+ private final BlockingQueue<InternalQueueEntry> queue = new
LinkedBlockingQueue<>();
+ private final AtomicInteger messageCount = new AtomicInteger(0);
+
+ /**
+ * Internal queue entry type used to discriminate between messages and
wakeup signals.
+ *
+ * This sealed interface ensures type safety when polling the queue.
+ */
+ private sealed interface InternalQueueEntry { }
+
+ /**
+ * Marker entry used to unblock threads waiting on {@link #poll(long)}
without delivering a message.
+ *
+ * Wakeup entries are drained during polling and do not contribute to the
message count.
+ */
+ private record WakeupMarker() implements InternalQueueEntry { }
+
+ private static final WakeupMarker WAKEUP = new WakeupMarker();
+
+ /**
+ * A queue entry that contains a message and its associated future.
+ */
+ private static final class MessageEntry implements QueueEntry,
InternalQueueEntry {
+ private final CompletableFuture<RaftMessage> future = new
CompletableFuture<>();
+ private final RaftMessage message;
+
+ MessageEntry(RaftMessage message) {
+ this.message = message;
+ }
+
@Override
- public int correlationId() {
- return 0;
+ public RaftMessage message() {
+ return message;
}
@Override
- public ApiMessage data() {
- return null;
+ public CompletableFuture<RaftMessage> future() {
+ return future;
}
- };
- private final BlockingQueue<RaftMessage> queue = new
LinkedBlockingQueue<>();
- private final AtomicInteger size = new AtomicInteger(0);
+ @Override
+ public String toString() {
+ return String.format(
+ "MessageEntry(message=%s, future.isDone=%s)",
+ message,
+ future.isDone()
+ );
+ }
+ }
@Override
- public RaftMessage poll(long timeoutMs) {
+ public Optional<QueueEntry> poll(long timeoutMs) {
Review Comment:
JDK uses timeout for the parameter name with similar semantic.
> I think making mistakes with units can be quite common and is something to
be actively avoided.
Yes, Kafka traditionally has not used TimeUnit in its APIs. The modern way
to represent time is with Duration but that is beyond the scope of this PR.
--
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]
