How to avoid 'Mutiny had to drop the following exception'

[ennishol]

Hello!

if I run the following test, it is green but Mutiny logs [-- Mutiny had to drop the following exception --] which is thrown by Uni<String> a

Is there a way to properly cancel other Unis if the first one was already received, so that no dropped exceptions are reported? It seems that it happens only if the unis run on an executor.

Thanks!

The test:

import io.smallrye.mutiny.Uni;
import io.smallrye.mutiny.infrastructure.Infrastructure;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

class UniFirstTest {

    @Test
    void uni() throws Exception {
        var bean = new Bean();
        Uni<String> a = Uni.createFrom().item(() -> bean.getValue("f")).runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
        Uni<String> b = Uni.createFrom().item(() -> bean.getValue("test")).runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
        var subscribe = Uni.join().first(a, b).withItem().subscribe();
        assertThat(subscribe.asCompletionStage().get(10, TimeUnit.SECONDS)).isEqualTo("testtest");
    }

    private static class Bean {

        public String getValue(String test) {
            if (test == null || test.length() < 3) {
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                throw new IllegalArgumentException("too short");
            }
            return test + test;
        }
    }
}

[jponge]

UniJoinFirst actually cancels the remaining Uni instances when the first one emits an item.

What you are seeing here is actually normal, the upstream Uni is running (blocking) code in getValue. So ultimately an exception is thrown after cancellation, and this is still being reported because creating from an item should not throw, and because the code in getValue has no way to cancel (Mutiny cannot automagically cancel code).

If you want to run some code that can either emit an item or report an exception, you might use Uni.createFrom().emitter(...).

[Wytzepakito]

Hey,
I have a similar situation and I don't understand how I would solve this using an emitter and still have the nice feature of first() to return a result as soon as the first Uni finishes.

I have two api calls and am sure only either of the calls will succeed or both will fail. If both fail I'm interested to see the underlying logs.
I could use an emitter on both calls and capture the errors and return some item that contains an Optional<value> and Optional <throwable> and then write logic to spew errors in case both calls report exceptions.

But since both calls now emit an item, I have to always await both calls even though the first call might have technically succeeded.

I hope it's clear, I can also write an example!

Thanks a lot!

[Wytzepakito]

Okay somehow rewriting the test to:

    @Test
    void uni() throws Exception {
        var bean = new Bean();
        Uni<String> a = Uni.createFrom().emitter(emitter -> {
            try {
                String res = bean.getValue("t");
                emitter.complete(res);
            } catch (Exception e) {
                emitter.fail(e);
            }
        });
        Uni<String> a2 = a.runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
        Uni<String> b = Uni.createFrom().emitter(emitter -> {
            try {
                String res = bean.getValue("test");
                emitter.complete(res);
            } catch (Exception e) {
                emitter.fail(e);
            }
        });
        Uni<String> b2 = b.runSubscriptionOn(Infrastructure.getDefaultWorkerPool());
        var subscribe = Uni.join().first(a2, b2).withItem().subscribe();
        assertTrue(subscribe.asCompletionStage().get(10, TimeUnit.SECONDS).equals("testtest"));
    }

Does work but I don't fully understand why it works. I would expect Mutiny to still log the exception.

[Wytzepakito]

Looking deeper into it I think I have an idea now...

Thanks a lot for the rubber-ducking!

[jponge]

Just for the record: those exceptions are being logged / reported to avoid loosing some information. A typical example is when a failure is forwarded to an operator that has already completed and the original emitter didn't have time to be cancelled.

[Wytzepakito]

If you're okay with checking my thought process... When I use my own emitter we basically avoid the code that's normally executed here

because when using Uni.createFrom().item() we basically create an emitter which when executing onTermination the isCancelled field is checked and this then executes this Runnable, correct?

[jponge]

Well you use a join first strategy, so one part of the pipeline may forward an item or failure that will be discarded.