Unit tests fail for MutinyEmitter with ProducerRecord

[enrico-niccoli]

Describe the bug

Hi,

I have defined a MutinyEmitter to send data to Kafka. Since I need to dynamically set the topic name and to handle possible errors, my Emitter is defined as follows:

@Inject
@Channel("message")
MutinyEmitter<ProducerRecord<String, MyObject>> streamsEmitter;

@POST
public Uni<Response> sendMyObject(MyObject obj) {

    return streamsEmitter.send(new ProducerRecord<>(obj.getTopicName(), obj))
            .map(x -> Response.ok("Message sent").build())
            .onFailure().recoverWithItem(Response.serverError().entity("Message not sent correctly").build());

}

This works as expected but executing tests defined as described in doc returns org.awaitility.core.ConditionTimeoutException: com.package.MyTestClass expected the predicate to return but it returned for input of <[]> within 10 seconds.

My test class is defined as follows:

@QuarkusTest
@QuarkusTestResource(KafkaTestResourceLifecycleManager.class)
class MyTestClass {

    @Inject @Any
    InMemoryConnector connector;
    @Inject
    MyResourceClass resource;

    @Test
    void testMessageSender() {

        InMemorySink<MyObject> messages = connector.sink("message");

        MyObject obj = new MyObject();
        obj.setTopicName("test");

        resource.sendMyObject(obj);

        await().<List<? extends Message<MyObject>>>until(messages::received, t -> t.size() == 1);

        final var received = messages.received().get(0).getPayload();
        assertNotNull(received);
    }
}

I suspect that this behaviour could be caused by the use of MutinyEmitter<ProducerRecord<String, MyObject>> instead of MutinyEmitter<MyObject, but I don't know how to fix it.

Thanks,
Enrico

Expected behavior

The in-memory channel receives correctly the data and test pass

Actual behavior

Test failed with org.awaitility.core.ConditionTimeoutException: com.package.MyTestClass expected the predicate to return but it returned for input of <[]> within 10 seconds.

How to Reproduce?

No response

Output of uname -a or ver

Linux asset-00287 6.2.0-36-generic #37~22.04.1-Ubuntu SMP PREEMPT_DYNAMIC Mon Oct 9 15:34:04 UTC 2 x86_64 x86_64 x86_64 GNU/Linux

Output of java -version

openjdk version "17.0.8"

Quarkus version or git rev

3.2.6.Final

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

Apache Maven 3.11.0

Additional information

No response

[quarkus-bot]

/cc @cescoffier (mutiny), @jponge (mutiny)

[jponge]

/cc @ozangunalp (Reactive Messaging)

[ozangunalp]

There are multiple problems with your test code:

1. You call the sendMyObject method which returns a Uni and it is never subscribed to. So the message is never dispatched. You can fix that by adding .await().indefinitely();
2. The MutinyEmitter generic type ProducerRecord<String, MyObject> doesn't match the InMemorySink type of MyObject. Note that InMemoryConnector is agnostic of your Kafka type. So the sink received will accumulate objects of type Message<ProducerRecord<String, MyObject>>.

[ozangunalp]

I am converting this issue into a discussion