Getting "Result is already complete" exceptions when publishing messages over event bus

[guss77]

Version

Vert.x 4.4.6

Context

I have a set of verticles in a cluster (each on their own server), using Hazelcast as the cluster implementation. There's a singleton class that does this in its constructor:

eventBus.<JsonObject>consumer(REGISTRY_ADDRESS, this::receivedRemoteUpdate);
this.updatePublisher = eventBus.<JsonObject>publisher(REGISTRY_ADDRESS);

The receivedRemoteUpdate() method doesn't do anything interesting in terms of Vert.x event bus, it just accepts the message and parses it:

public void receivedRemoteUpdate(Message<JsonObject> message) {
	var update = message.body().mapTo(MyJSONBean.class);
	// do something with the bean data
}

When there is something to publish about, the code does this:

updatePublisher.write(JsonObject.mapFrom(new MyJsonBean(somedata)))
.onSuccess(r -> log.info("Published data successuflly"))
.onFailure(t -> log.error("Error publishing data", t));

Sometimes (not always), after I see the info log line, I get this error in the logs:

2023-10-30 12:20:29.000 [vert.x-eventloop-thread-2] ERROR io.vertx.core.impl.ContextBase - Unhandled exception
2023-10-30 12:20:29.000 java.lang.IllegalStateException: Result is already complete
2023-10-30 12:20:29.000       at io.vertx.core.Promise.complete(Promise.java:78)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.OutboundDeliveryContext.written(OutboundDeliveryContext.java:85)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.EventBusImpl.sendLocally(EventBusImpl.java:345)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.EventBusImpl.sendOrPub(EventBusImpl.java:329)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.sendToNode(ClusteredEventBus.java:335)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.sendToNodes(ClusteredEventBus.java:346)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.clustered.ClusteredEventBus.lambda$sendOrPub$6(ClusteredEventBus.java:217)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:141)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.PromiseImpl.tryComplete(PromiseImpl.java:23)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.PromiseImpl.onSuccess(PromiseImpl.java:49)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.FutureImpl$ListenerArray.onSuccess(FutureImpl.java:262)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
2023-10-30 12:20:29.000       at io.vertx.core.impl.future.PromiseImpl.tryComplete(PromiseImpl.java:23)
2023-10-30 12:20:29.000       at io.vertx.core.spi.cluster.impl.DefaultNodeSelector.lambda$selectForPublish$1(DefaultNodeSelector.java:51)
2023-10-30 12:20:29.000       at io.vertx.core.spi.cluster.impl.selector.Selectors.withSelector(Selectors.java:55)
2023-10-30 12:20:29.000       at io.vertx.core.spi.cluster.impl.DefaultNodeSelector.selectForPublish(DefaultNodeSelector.java:50)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.clustered.Serializer$SerializerQueue$SerializedTask.process(Serializer.java:152)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.clustered.Serializer$SerializerQueue.checkPending(Serializer.java:99)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.clustered.Serializer$SerializerQueue.add(Serializer.java:119)
2023-10-30 12:20:29.000       at io.vertx.core.eventbus.impl.clustered.Serializer.lambda$queue$1(Serializer.java:67)
2023-10-30 12:20:29.000       at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:55)
2023-10-30 12:20:29.000       at io.vertx.core.impl.EventLoopContext.lambda$emit$1(EventLoopContext.java:62)
2023-10-30 12:20:29.000       at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:173)
2023-10-30 12:20:29.000       at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:166)
2023-10-30 12:20:29.000       at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
2023-10-30 12:20:29.000       at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:569)
2023-10-30 12:20:29.000       at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
2023-10-30 12:20:29.000       at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
2023-10-30 12:20:29.000       at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
2023-10-30 12:20:29.000       at java.base/java.lang.Thread.run(Thread.java:1583)

Do you have a reproducer?

I do not have a rerproducer

Extra

The code in question is running using the Amazon Corretto JDK 21 container on AWS Fargate.

[tsegismont]

This happens only when using a producer instead of invoking eventBus.publish(). In this case only, the outbound delivery context is created using a non-null write-promise.
And when the producer is of publish type, the OutboundDeliveryContext#written can be invoked several times.

I see two options:

1. replace writePromise.complete() and writePromise.fail(failure) with their tryXXX equivalent in OutboundDeliveryContext#written`, or
2. return an already succeeded future in MessageProducerImpl#write when the producer is of publish type

My preference goes to (2) because the information conveyed by the future result is not reliable in this case anyway.

Thoughts @vietj ?

[vietj]

we should define what means success or failure for publishing messages, perhaps it just means always success as you said with (2)

[tsegismont]

Actually, a combination of the two would be best

[tsegismont]

Fixed by 0b0e51b