Getting io.vertx.core.VertxException: Thread blocked

[PyAntony]

Hello,

I am in fact blocking the eventloop thread, but this is exactly what I want to accomplish. Here is what I am doing:

    @Incoming("copy-ready-topic")
    @Acknowledgment(Acknowledgment.Strategy.NONE)
    public Subscriber<IncomingKafkaRecord<String, PlayerCopyReady>> copyTopicConsumer() {
        return new BaseSubscriber<>() {
            @Override
            public void hookOnSubscribe(Subscription subscription) {
                request(1);
            }

            @Override
            public void hookOnNext(IncomingKafkaRecord<String, PlayerCopyReady> message) {
                boolean capacityIsFull;
                // insert payload in queue
                // `insert` returns true when payload was inserted
                do {
                    capacityIsFull = !keyFinderCache.insert(message);
                } while (capacityIsFull);

                request(1);
            }
        };
    }

The idea is simple, I am inserting all messages in a queue object in the hookOnNext method. This queue has a limit and when limit is reached insertion returns false (no insertion happens). Program does additional processing from this queue, the additional processing is irrelevant for this discussion, but sometimes this processing becomes slow, queue grows to limit and the do-while loop keeps looping for multiple seconds. As you can imagine I get the io.vertx.core.VertxException: Thread blocked exception printed from this consumer thread (program doesn't crash thankfully), but what the program is doing is exactly what I want to accomplish! I want to consume messages only while queue has capacity, program is working as expected. From here I have 2 questions:

1. The internal KafkaConsumer (from org.apache.kafka.clients.consumer) used by the framework does stop polling while I am not consuming more records correct?
2. Given that the program is doing what I want, should I just ignore the io.vertx.core.VertxException or is there something wrong in my logic?

[cescoffier]

Your logic is blocking so you need to run it on a worker thread. You can loop like this on the event loop. However, here a few other suggestions.

Also, if you do not request, the kafka connector will automatically pause the consumption and so, you may not have to do anything.

Finally, you delay some action, you can use vertx.setTimer(1000, x->...). This will not block the event loop.

[PyAntony]

how can I run only the hookOnNext method in a worker pool? or maybe schedule the hookOnNext method to be invoked every second inside the Subscriber? could you give examples please?

[cescoffier]

From where comes this method? This seems to be a custom method from an RX programming lib. We recommend using Mutiny, which is much simpler.
I guess this is just the onItem/onNext callback, but I do not know that method.

[PyAntony]

I found a solution following your advice ("if you do not request, the kafka connector will automatically pause the consumption and so, you may not have to do anything"). Basically my solution is:

// create a subscriber and assign it to a variable
@PostConstruct
void init() {
    copyReadySubscriber = new BaseSubscriber<>() {
        @Override
        public void hookOnNext(IncomingKafkaRecord<String, PlayerCopyReady> message) {
            // subscriber inserts payloads in cache (cache from this class or any other class)
            keyFinderCache.insert(message.getPayload());
        }
    };
}

// just return this subscriber here
@Incoming("copy-ready-topic")
@Acknowledgment(Acknowledgment.Strategy.NONE)
public Subscriber<IncomingKafkaRecord<String, PlayerCopyReady>> copyTopicConsumer() {
    return copyReadySubscriber;
}

// then you can use the subscriber reference anywhere to keep requesting. For example:
@Scheduled(
    every = "1s",
    concurrentExecution = SKIP
)
void pushCopyReadyRequests() {
    // any condition you want
    if (keyFinderCache.hasCapacity())
        copyReadySubscriber.request(batchNumber);
}

[PyAntony]

Do you have an extensive documentation for Mutiny? something more extensive than the current guides in the web site? (Something like Reactor: https://projectreactor.io/docs/core/release/reference/). I have tried Mutiny on few occasions but I always end up with Reactor out of frustration (could be because I started with Reactor in the first place so i am biased).

[cescoffier]

Did you check: https://smallrye.io/smallrye-mutiny/ ?

[PyAntony]

yes, I have. Don't feel it's extensive enough though but I need to go over the entire site. I am slowly moving towards Mutiny anyway given that it plays nicer with Quarkus. Thank you!

[cescoffier]

What are the topics you would like to see? We are planning a re-platforming (to mkdocs) and adding a few topics. So if there is anything you would like to see, that's the perfect time.

[PyAntony]

nothing in particular, just more detailed content in general, for example: https://smallrye.io/smallrye-reactive-messaging/3.15.0/getting-started/

I think it is important also to put everything in 1 single page also so it is easy to find things by searching strings. Opening tabs might be good for tutorials but not to search.

Also, a single page with an exhaustive list of all operations possible. For example: https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Flux.html

As a site note, I just realized Uni is not a Publisher! this is why Uni didn't play well with Multi.flatmap(...) (coming from Reactor this was a huge surprise)