4

I have an endpoint which pushes data to kafka. Now, I want to respond to the call with the appropiriate status code 2xx or 5xx in case of kafka write success or failure respectively. The code snippet is

@Path("/prices")
public class PriceResource {

    @Inject @Channel("price-create") Emitter<Double> priceEmitter;

    @POST
    @Consumes(MediaType.TEXT_PLAIN)
    public void addPrice(Double price) {
        priceEmitter.send(Message.of(price)
            .withAck(() -> {
                // Called when the message is acked
                return CompletableFuture.completedFuture(null);
            })
            .withNack(throwable -> {
                // Called when the message is nacked
                return CompletableFuture.completedFuture(null);
            }));
         // return appropriate response
          return Response
    }
}

Now the issue is that the endpoint is responding with the status code before ack or nack callback is executed. Also tried the sendAndAwait method of MutinyEmitter but that method returns void. So there is no way to know if the message is acked or nacked.

rightjoin
  • 77
  • 8

1 Answers1

4

The best way here is to chain async operations, like this:

@POST
@Consumes(MediaType.TEXT_PLAIN)
public Uni<Response> addPrice(Double price) {
    return Uni.createFrom().completionStage(priceEmitter.send(price))
            .onItem().transform(ignored -> Response.ok().entity("foo").build())
            .onFailure().recoverWithItem(Response.serverError().build());
}

If you want to work with sync code (I do not recommend):

@Blocking
@POST
@Consumes(MediaType.TEXT_PLAIN)
public Response addPrice(Double price) {
    try {
        Uni.createFrom().completionStage(priceEmitter.send(price))
                .await().indefinitely();

        return Response.ok().entity("foo").build();
    } catch (Exception e) {
        return Response.serverError().build();
    }
}

The .await().indefinitely() will throw an exception in case of the Uni emitting a failure.

You also have the option to work directly with the CompletionStage returned by the emitter without converting it to a Uni, but remember Quarkus chose Mutiny as its default reactive framework.

Vinicius
  • 1,060
  • 1
  • 12
  • 21
  • @Vincius thanks. One question though. What if instead returning Uni, I just want to return a String. How do I get the value on fail/success from Uni. – rightjoin Sep 30 '21 at 08:06
  • 1
    You can use Uni#await and annotate the method with @Blocking to get a sync response but I do not recommend this since Emitter#send is async in nature. You can easily return the String using Response.ok().entity(...). – Vinicius Sep 30 '21 at 10:08