Skip to content

make sure after resume we just deliver one message at one moment #5525

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: 4.x
Choose a base branch
from
Open

make sure after resume we just deliver one message at one moment #5525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e93ab84
make sure the delivery of the message has no race condition
hadadil Mar 14, 2025
b408a6d
removing the resume as that is basically wiping out pending msgs duri…
hadadil Mar 14, 2025
ebaab76
avoid StackOverflow and do return with false when msg discarded so me…
hadadil Mar 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 17 additions & 19 deletions src/main/java/io/vertx/core/eventbus/impl/MessageConsumerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -138,28 +138,26 @@ protected boolean doReceive(Message<T> message) {
if (handler == null) {
return false;
}
if (demand == 0L) {
if (pending.size() < maxBufferedMessages) {
pending.add(message);
if (pending.size() < maxBufferedMessages) {
pending.add(message);
if (demand == 0) {
return true;
} else {
discard(message);
if (discardHandler != null) {
discardHandler.handle(message);
} else {
log.warn("Discarding message as more than " + maxBufferedMessages + " buffered in paused consumer. address: " + address);
if (demand != Long.MAX_VALUE) {
demand--;
}
}
return true;
} else {
if (pending.size() > 0) {
pending.add(message);
message = pending.poll();
theHandler = handler;
}
if (demand != Long.MAX_VALUE) {
demand--;
} else {
discard(message);
if (discardHandler != null) {
discardHandler.handle(message);
} else {
String pause = demand == 0 ? "paused" : "NOT paused";
log.warn("Discarding message as more than " + maxBufferedMessages + " buffered in " + pause + " consumer. address: " + address);
}
theHandler = handler;
return false;
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this was also not 100% right, when we do discard msg we should return with false to have that in the metrics

}
}
deliver(theHandler, message);
Expand All @@ -171,7 +169,7 @@ protected void dispatch(Message<T> msg, ContextInternal context, Handler<Message
if (handler == null) {
throw new NullPointerException();
}
context.emit(msg, handler);
context.dispatch(msg, handler);
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this was reverted as it had a side effect:
after resume we were received many messages even we paused again during the first message processing.

}

private void deliver(Handler<Message<T>> theHandler, Message<T> message) {
Expand All @@ -183,8 +181,8 @@ private void deliver(Handler<Message<T>> theHandler, Message<T> message) {

private synchronized void checkNextTick() {
// Check if there are more pending messages in the queue that can be processed next time around
if (!pending.isEmpty() && demand > 0L) {
context.nettyEventLoop().execute(() -> {
if (demand > 0L && !pending.isEmpty()) {
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I flip here so the demand evaulated first

context.execute(__ -> {
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

here we do execute the resume/fetch from the context thread so we do not have situations when we call resume and the same time we do receive a msg and we get 2msgs (on on the context thread and one on eventloop thread)

Message<T> message;
Handler<Message<T>> theHandler;
synchronized (MessageConsumerImpl.this) {
Expand Down
1 change: 0 additions & 1 deletion src/test/java/io/vertx/core/eventbus/LocalEventBusTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1401,7 +1401,6 @@ public void testUnregisterConsumerDiscardPendingMessages() {
eb.send(ADDRESS1, "val1");
Context ctx = Vertx.currentContext();
ctx.runOnContext(v -> {
consumer.resume();
Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this was in the past executed on eventloop thread on the context... now this is executed on the context thread (in case we area lready on it it will not schedule new job, in case we are on another thread it will schedule an async job.
in short:
now this is executed here so the logic below was not anymore fine as this call wipe out sync the pending msgs

((MessageConsumerImpl<?>) consumer).discardHandler(discarded -> {
assertEquals("val1", discarded.body());
testComplete();
Expand Down
53 changes: 42 additions & 11 deletions src/test/java/io/vertx/core/eventbus/MessageConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,33 +7,41 @@
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;

public class MessageConsumerTest extends VertxTestBase {


@Test
public void testMessageConsumptionStayOnWorkerThreadAfterResume() throws Exception {
TestVerticle verticle = new TestVerticle(2);
public void testMessageConsumptionStayOnWorkerThreadAfterResumeAndOnlyDispatchOneMessageAtOneMoment() throws Exception {
int numberOfExpectedMessages = 10;
TestVerticle verticle = new TestVerticle(numberOfExpectedMessages);
EchoVerticle echoVerticle = new EchoVerticle();
Future<String> deployVerticle = vertx.deployVerticle(verticle, new DeploymentOptions().setThreadingModel(ThreadingModel.WORKER));
Future<String> deployEchoVerticle = vertx.deployVerticle(echoVerticle, new DeploymentOptions().setThreadingModel(ThreadingModel.WORKER));

CountDownLatch startLatch = new CountDownLatch(1);
deployVerticle.onComplete(onSuccess(cf -> startLatch.countDown()));
Future.all(deployVerticle, deployEchoVerticle)
.onComplete(onSuccess(cf -> startLatch.countDown()));
awaitLatch(startLatch);

vertx.eventBus().send("testAddress", "message1");
vertx.eventBus().send("testAddress", "message2");
for (int i = 1; i <= numberOfExpectedMessages; i++) {
vertx.eventBus().send("testAddress", "message" + i);
}

awaitLatch(verticle.msgLatch);

assertEquals(2, verticle.messageArrivedOnWorkerThread.size());
assertTrue("message1 should be processed on worker thread", verticle.messageArrivedOnWorkerThread.get("message1"));
assertTrue("message2 should be processed on worker thread", verticle.messageArrivedOnWorkerThread.get("message2"));
assertEquals(numberOfExpectedMessages, verticle.messageArrivedOnWorkerThread.size());
for (int i = 1; i <= numberOfExpectedMessages; i++) {
assertTrue("message" + i + " should be processed on worker thread", verticle.messageArrivedOnWorkerThread.get("message" + i));
}
}


private static class TestVerticle extends AbstractVerticle {

private final CountDownLatch msgLatch;
private final AtomicBoolean messageProcessingOngoing = new AtomicBoolean();

private final Map<String, Boolean> messageArrivedOnWorkerThread = new HashMap<>();

Expand All @@ -51,11 +59,34 @@ private void handleMessages(MessageConsumer<String> consumer) {
consumer.handler(msg -> {
consumer.pause();
messageArrivedOnWorkerThread.putIfAbsent(msg.body(), Context.isOnWorkerThread());
msgLatch.countDown();
vertx.setTimer(20, id -> {
consumer.resume();
if (messageProcessingOngoing.compareAndSet(false, true)) {
msgLatch.countDown();
} else {
System.err.println("Received message while processing another message");
}
vertx.eventBus().request("echoAddress", 20)
.onComplete(ar -> {
messageProcessingOngoing.set(false);
consumer.resume();
});
});
}
}

private static class EchoVerticle extends AbstractVerticle {
@Override
public void start() {
MessageConsumer<Integer> consumer = vertx.eventBus().localConsumer("echoAddress");
handleMessages(consumer);
}

private void handleMessages(MessageConsumer<Integer> consumer) {
consumer.handler(msg -> {
vertx.setTimer(msg.body(), id -> {
msg.reply(msg.body());
});
});
}
}

}
Loading