Reactive Messaging: Tombstone-messages no longer accepted

[wicksim]

Describe the bug

Since Quarkus 2.14.0.Final, I can no longer send messages with null-value over the Emitter. I get this exception:

2023-02-08 14:21:13,807 ERROR [io.sma.rea.mes.kafka] (Quarkus Main Thread) SRMSG18207: Unable to dispatch message to Kafka: java.lang.IllegalArgumentException: `payload` must not be `null`
	at org.eclipse.microprofile.reactive.messaging.Message.of(Message.java:304)
	at org.eclipse.microprofile.reactive.messaging.Message.addMetadata(Message.java:532)
	at io.quarkus.opentelemetry.runtime.tracing.intrumentation.reactivemessaging.ReactiveMessagingTracingDecorator.lambda$decorate$1(ReactiveMessagingTracingDecorator.java:75)
	at io.smallrye.context.impl.wrappers.SlowContextualFunction.apply(SlowContextualFunction.java:21)
	at io.smallrye.mutiny.operators.multi.MultiMapOp$MapProcessor.onItem(MultiMapOp.java:42)
	at io.smallrye.mutiny.operators.multi.MultiOperatorProcessor.onItem(MultiOperatorProcessor.java:100)
	at io.smallrye.mutiny.operators.multi.builders.BufferItemMultiEmitter.drain(BufferItemMultiEmitter.java:118)
	at io.smallrye.mutiny.operators.multi.builders.BufferItemMultiEmitter.emit(BufferItemMultiEmitter.java:34)
	at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.onItem(SerializedMultiEmitter.java:51)
	at io.smallrye.mutiny.operators.multi.builders.SerializedMultiEmitter.emit(SerializedMultiEmitter.java:141)
	at io.smallrye.reactive.messaging.providers.extension.ThrowingEmitter.emit(ThrowingEmitter.java:63)
	at io.smallrye.reactive.messaging.providers.extension.AbstractEmitter.emit(AbstractEmitter.java:164)
	at io.smallrye.reactive.messaging.providers.extension.EmitterImpl.send(EmitterImpl.java:47)
	at org.acme.TestEmitter.emit(TestEmitter.java:46)
	at org.acme.TestEmitter.atStartup(TestEmitter.java:29)
	at org.acme.TestEmitter_Observer_atStartup_0f975a6ffd0cd0e4d2a55a386da871634c38d3e9.notify(Unknown Source)
	at io.quarkus.arc.impl.EventImpl$Notifier.notifyObservers(EventImpl.java:328)
	at io.quarkus.arc.impl.EventImpl$Notifier.notify(EventImpl.java:310)
	at io.quarkus.arc.impl.EventImpl.fire(EventImpl.java:78)
	at io.quarkus.arc.runtime.ArcRecorder.fireLifecycleEvent(ArcRecorder.java:131)
	at io.quarkus.arc.runtime.ArcRecorder.handleLifecycleEvents(ArcRecorder.java:100)
	at io.quarkus.deployment.steps.LifecycleEventsBuildStep$startupEvent1144526294.deploy_0(Unknown Source)

As far as I can see, this happens since ReactiveMessagingTracingDecorator was introduced. This class tries to add a header and therefore rebuilds the message.

Expected behavior

It should be possible to send messages with null-value (tombstone-messages).

Actual behavior

It is not possible to send tombstone-messages.

How to Reproduce?

Reproducer:
empty-value-reproducer.zip

1. Execute reproducer
2. Wait a few seconds
3. See exception

Output of uname -a or ver

Microsoft Windows [Version 10.0.19044.2006]

Output of java -version

OpenJDK Runtime Environment Corretto-17.0.4.9.1 (build 17.0.4.1+9-LTS)

GraalVM version (if different from Java)

No response

Quarkus version or git rev

2.16.1.Final

Build tool (ie. output of mvnw --version or gradlew --version)

Apache Maven 3.8.6

Additional information

No response

[quarkus-bot]

/cc @cescoffier (reactive-messaging), @ozangunalp (reactive-messaging)

[cescoffier]

Ourf, nice catch....

@ozangunalp can you have a look?

[ozangunalp]

There is an immediate workaround to use message payload as io.smallrye.reactive.messaging.kafka.Record or ProducerRecord with null value.

@cescoffier maybe the definitive fix is to remove the payload null check in Message methods. WDYT ?

[cescoffier]

@ozangunalp they come from the spec...

[ozangunalp]

I can't find it in the spec. The TCK doesn't check for it.
The only payload null check is in the Emitter#send(T payload).

[cescoffier]

Hum... ok, then, let's remove the checks.
The Message class comes from the spec; that's why I was pretty sure it was mandated.

[wicksim]

@ozangunalp Do you have any expectation in which future release this will be fixed? I've seen the proposed workaround, but we are using Message in a lot of code and to change all that code for a workaround is really unpleasant.

[ozangunalp]

I can understand that. I need to do more checks before going through with the change. I'll get back to you here at the end of the day.

[gsmet]

@ozangunalp let me know if there is something I should backport (or if we need a PR specific to 2.16). Thanks.

[cescoffier]

@ozangunalp wasn't that one fixed?

[ozangunalp]

On main yes. I need to see what we include to the backport release.

[wicksim]

@ozangunalp The referenced PR was closed, because the risk was considered too high. What does this mean regarding this issue? Will it not be fixed until release 3?

[cescoffier]

For sure, it will be in Quarkus 3.

[ozangunalp]

It is already merged into the Quarkus main branch, so will be in Quarkus 3.
Unless there is a specific need I won't be backporting this to 2.16.

[wicksim]

There is a need from our side, because we would be stuck on 2.13 until 3 is out. But of course I'm not sure if this need is big enough if we are the only ones... 😉

[humbertosales]

The workaround working in 2.16 version. :-)

import javax.enterprise.context.ApplicationScoped;
import javax.inject.Inject;

import org.eclipse.microprofile.reactive.messaging.Channel;
import org.eclipse.microprofile.reactive.messaging.Emitter;
import org.eclipse.microprofile.reactive.messaging.Message;

import io.smallrye.reactive.messaging.kafka.Record;
import io.smallrye.reactive.messaging.kafka.api.OutgoingKafkaRecordMetadata;

@ApplicationScoped
public class EmitterTester<K, V>  {
	@Inject
	@Channel("emitterTest")
	protected Emitter<Record<K,V>> emitterToGenericPersistence;

	public void send(String topic, K key, V value) {

        OutgoingKafkaRecordMetadata<Object> metadata = OutgoingKafkaRecordMetadata.builder()
                .withTopic(topic)
                .withKey(key) //JsonSerializer
                .build();

        
        //https://github.com/quarkusio/quarkus/issues/31003
        //payload not null
        Message<Record<K,V>> message = Message.of(Record.of(key, value)).addMetadata(metadata);
        
        emitterToGenericPersistence.send(message);
        
	}

}