Mutiny - Kafka writes happening sequentially

I am new to Quarkus. I am trying to write a REST endpoint using quarkus reactive that receives an input, does some validation, transforms the input to a list and then writes a message to kafka. My understanding was converting everything to Uni/Multi, would result in the execution happening on the I/O thread in async manner. In, the intelliJ logs, I could see that the code is getting executed in a sequential manner in the executor thread. The kafka write happens in its own network thread sequentially, which is increasing latency.

@POST
    @Consumes(MediaType.APPLICATION_JSON)
    @Produces(MediaType.APPLICATION_JSON)
    public Multi<OutputSample> send(InputSample inputSample) {
        ObjectMapper mapper = new ObjectMapper();

       //deflateMessage() converts input to a list of inputSample
        Multi<InputSample> keys = Multi.createFrom().item(inputSample)
                .onItem().transformToMulti(array -> Multi.createFrom().iterable(deflateMessage.deflateMessage(array)))
                .concatenate();

     
        return keys.onItem().transformToUniAndMerge(payload -> {
            try {
                return producer.writeToKafka(payload, mapper);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            return null;
        });
    }

@Inject
@Channel("write")
    Emitter<String> emitter;

Uni<OutputSample> writeToKafka(InputSample kafkaPayload, ObjectMapper mapper) throws JsonProcessingException {
        
        String inputSampleJson = mapper.writeValueAsString(kafkaPayload);
        return Uni.createFrom().completionStage(emitter.send(inputSampleJson))
                .onItem().transform(ignored -> new OutputSample("id", 200, "OK"))
                .onFailure().recoverWithItem(new OutputSample("id", 500, "INTERNAL_SERVER_ERROR"));
    }

I have been on it for a couple of days. Not sure if doing anything wrong. Any help would be appreciated. Thanks

Solution

Multi is intended to be used when you have a source that emits items continuously until it emits a completion event, which is not your case.

From Mutiny docs:

A Multi represents a stream of data. A stream can emit 0, 1, n, or an infinite number of items.

You will rarely create instances of Multi yourself but instead use a reactive client that exposes a Mutiny API.

What you are looking for is a Uni<List<OutputSample>> because your API you return 1 and only 1 item with the complete result list.

So what you need is to send each message to Kafka without immediately waiting for their return but collecting the generated Unis and then collecting it to a single Uni.

@POST
public Uni<List<OutputSample>> send(InputSample inputSample) {
    // This could be injected directly inside your producer
    ObjectMapper mapper = new ObjectMapper();

    // Send each item to Kafka and collect resulting Unis
    List<Uni<OutputSample>> uniList = deflateMessage(inputSample).stream()
            .map(input -> producer.writeToKafka(input, mapper))
            .collect(Collectors.toList());

    // Transform a list of Unis to a single Uni of a list
    @SuppressWarnings("unchecked") // Mutiny API fault...
    Uni<List<OutputSample>> result = Uni.combine().all().unis(uniList)
            .combinedWith(list -> (List<OutputSample>) list);

    return result;
}