How to mock/test nack with InMemorySink?

[vdilchev]

Hi there,

I am trying to test the handling of NACK in my Outgoing Channel Emitter.
So I am injecting and using an Emitter like that:

#pseudocode
@Channel("my-out-channel") Emitter<MyPayload> emitter;
# in a method called with payload
emitter.send(Message.of(payload).withNack(t -> CompletableFuture.failedFuture(wrap(t)));

Could someone suggest a way to mock/simulate the Connector (InMemoryConnector or a mock wired by other means) nacking the Message so I can test the handling of the nack?
Also, what will happen with the CompletableFuture returned by the nack-function?

It is a quarkus project with Kafka, JUnit and Mockito.

Thanks in advance.

[lMysticl]

InMemorySink cannot simulate a connector-side nack. Its implementation records every received message and then immediately calls m.ack(). It does not expose an operation that makes the sink reject a message.

You can still unit-test the nack callback by retrieving the emitted message and invoking it directly:

InMemorySink<MyPayload> sink = connector.sink("my-out-channel");

await().untilAsserted(() ->
        assertThat(sink.received()).hasSize(1)
);

Message<MyPayload> message = sink.received().get(0);
Throwable failure = new IllegalStateException("simulated broker failure");

CompletionStage<Void> nackResult = message.nack(failure);

assertThatThrownBy(() -> nackResult.toCompletableFuture().join())
        .isInstanceOf(CompletionException.class)
        .hasCauseInstanceOf(MyWrappedException.class);

This tests the callback registered through withNack, but it is not a full connector test: the in-memory sink has already acknowledged the message. For connector-level behavior, you need a small test outgoing connector/subscriber that calls message.nack(failure) instead of using InMemoryConnector.

The CompletionStage returned by your nack function becomes the result of Message.nack(...). If the callback returns CompletableFuture.failedFuture(wrap(t)), then message.nack(t) completes exceptionally with the wrapped exception.

Also note that Emitter.send(Message) returns void, so that overload does not expose the nack result to the sender. Emitter.send(payload) is the overload that returns a CompletionStage tied to acknowledgement or negative acknowledgement.

The behavior above follows directly from the current InMemorySinkImpl and EmitterImpl implementations.