"Blocking operation on a IO thread" problem in 0.23 - how to use kafka messaging in Quarkus properly nowadays?

[Sergey Bezrukov]

Hi,

our typical scenario is to read messages from Kafka, process it synchronously (it usually means we commit some data to database), then manually commit offset, so we can be sure that message was processed and we don't lost any data, if offest was commited, like this:

@Inject
SomeService someService; // has transactional method and EntityManager injected

@Incoming("some-topic")
processMessage(ReceivedKafkaMessage<String, byte[]> input){

someService.processData(input.getPayload); // process and save data to DB

input.ack();


}

But, after ugrading Quarkus to 0.23 we found that

io.quarkus.hibernate.orm.runtime.entitymanager.TransactionScopedEntityManager now contains checkBlocking() method, which leads to following exception in our code:


You have attempted to perform a blocking operation on a IO thread. This is not allowed, as blocking the IO thread will cause major performance issues with your application. If you want to perform blocking EntityManager operations make sure you are doing it from a worker thread.: java.lang.IllegalStateException: You have attempted to perform a blocking operation on a IO thread. This is not allowed, as blocking the IO thread will cause major performance issues with your application. If you want to perform blocking EntityManager operations make sure you are doing it from a worker thread.



I also found that algorithm, which detects is current thread is IOThread or not, was changed in master after 0.23, so, basically I have 2 questions:

1. Is thread, which handle @Incoming in our code, will still detected as "IOThread" after alorithm change will be released?
2. If so, how should we handle messages from Kafka, if we really need "synchronous" processing here, from message receiving till acknowledgment via JPA operations?

Thanks
--
Sergey 

[clement escoffier]

Hello,

To answer your questions:

1. - yes, the Kafka connector uses the same event loops as the HTTP layer, so they are IO Thread

2. - for the time being (because I’m working on a support of “blocking” in reactive messaging itself) you need to delegate to another thread, most probably to a single thread executor so you preserve the ordering. What happens in the executor can be synchronous and blocking.

Clement

--
You received this message because you are subscribed to the Google Groups "Quarkus Development mailing list" group.
To unsubscribe from this group and stop receiving emails from it, send an email to quarkus-dev...@googlegroups.com.
To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/a437ca9b-a37c-43e9-9608-94b38860b604%40googlegroups.com.

[Sergey Bezrukov]

Ok, thank you for the explanation.

--
Sergey Bezrukov
Cell: +7(903)790-93-34
Skype: sergey.bezrukov

[Luca Masini]

Well I think we have a problem in streams, in my Incoming:

@Incoming("ordini")
public void updateDevices(String deviceKeyJson) {
}

I can notice this thread running the incoming message:

2019-10-07 09:05:20,375 INFO [xxxxxx.StreamResource] (vert.x-eventloop-thread-1) ********************************** SSE sent to all deviceKey: {"property1":"999910763C000003071019090519","property1":486,"property1":"D","property1":"MONITOR_VASSOIO"}

Why don't the framework take care of this ?

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAD1V%3DtQQDVBJzEqfJ2DZDMNt%2Bi-8uW1krE0nG9Vu4UhkHMy9rA%40mail.gmail.com.

--

****************************************
http://www.lucamasini.net
http://twitter.com/lmasini
http://www.linkedin.com/pub/luca-masini/7/10/2b9
****************************************

[Stuart Douglas]

I think we need some way for reactive messaging to easily delegate to a worker thread, with both a global switch and a method level annotation to control which thread it runs it.

Running directly in the IO thread gives maximum performance, but it does not help much if you want to do blocking operations such as using JPA to write to the DB. We should make it easy to select the appropriate model.

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAG%3D-wFOn52GffuzaXfHiGDmEYG0YSfBWAzzQv2EW58FzpVP%3DfQ%40mail.gmail.com.

[Stephane Epardaud]

Agreed.

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAAoo%3Dc5WTc8NGK17TDSQJ%2BKN4iF92mPzr0WOSb8ttKp8KXG5RQ%40mail.gmail.com.

--

Stéphane Épardaud

[Ken Finnigan]

I think Clement, and the spec, had been discussing something like @Blocking to indicate that the operation is a blocking one

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAKU9E9sNAK8VBkBReavETVT_ag_rqK3EUn0FuP%2BpkDh76MWNyQ%40mail.gmail.com.

[David Lloyd]

Hopefully we could also infer whether an operation is blocking based
on what it injects/uses? Which implies that something like a
`@Blocking` would need a boolean parameter to allow overriding the
heuristic.

> To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAKeeVe7C4oh%2BQWC_RcQkdjOPFhnhjSzxRGsO-MLABn%2Br1sO7JA%40mail.gmail.com.



--
- DML

[Bruno Baptista]

Hi Clement,

Is the " Blocking" operation now supported?

Where can I track that work?

Cheers

Bruno Baptista
https://twitter.com/brunobat_

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/974DBC5D-9F05-4BF0-A480-61C1D44B72D2%40gmail.com.

[Ibrahim abed al aziz]

It is supported in Quarkus 1.5. You need to add @Blocking annotation to your method. 

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/b5080529-abf7-7a82-8ca8-bca1ba3d3b52%40gmail.com.

[clement escoffier]

Hello,

I've created a quickstart to demonstrate the feature:

https://github.com/quarkusio/quarkus-quickstarts/blob/f494722f558d282e6612edff1c5f7427788b4b68/kafka-panache-quickstart/src/main/java/org/acme/panache/PriceStorage.java

Clement

[Željko Trogrlić]

Note that if you want to use Message, like in the original example, using @Blocking is not that straightforward: https://github.com/smallrye/smallrye-reactive-messaging/issues/588

On Monday, June 8, 2020 at 8:32:44 PM UTC+2, clement escoffier wrote:

Hello,

I've created a quickstart to demonstrate the feature:

https://github.com/quarkusio/quarkus-quickstarts/blob/f494722f558d282e6612edff1c5f7427788b4b68/kafka-panache-quickstart/src/main/java/org/acme/panache/PriceStorage.java

Clement

Clement

To unsubscribe from this group and stop receiving emails from it, send an email to quark...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/a437ca9b-a37c-43e9-9608-94b38860b604%40googlegroups.com.

--
You received this message because you are subscribed to the Google Groups "Quarkus Development mailing list" group.

To unsubscribe from this group and stop receiving emails from it, send an email to quark...@googlegroups.com.

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/974DBC5D-9F05-4BF0-A480-61C1D44B72D2%40gmail.com.

--
You received this message because you are subscribed to the Google Groups "Quarkus Development mailing list" group.

To unsubscribe from this group and stop receiving emails from it, send an email to quark...@googlegroups.com.

[Bruno Baptista]

Thanks very much Clement!

The config is a bit thin. What will be the default config when you use the stock @Blocking?

How many concurrent requests will handled simultaneously, by default?

Cheers

Bruno Baptista
https://twitter.com/brunobat_

[clement escoffier]

Hello,

There are a bit more details on https://smallrye.io/smallrye-reactive-messaging/smallrye-reactive-messaging/2/advanced/blocking.html.

The default pool, in Quarkus, it can handle 20 tasks simultaneously. 

Clement

[Emmanuel Bernard]

I think Stuart discussed moving it up to 200  (with scale down capability), as 20 was a bit narrow in some situations.

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAKW6fidrRO0dTUno%2BDho6YJdPe9ZNNx%2Bin%3D8PFijQj2_M7mQ6Q%40mail.gmail.com.

[clement escoffier]

Hello,

I still see this:

/**
 * The size of the worker thread pool.
 */
@ConfigItem(defaultValue = "20")
public int workerPoolSize;

But, it can be configured. 

Clement

[Paul Carter-Brown]

The change to a default of 200 was done on the executor pool not the worker pool. The executor pool is the one that vertx eventloop threads hands work over to for processing incoming server requests. IIRC the worker pool is for async, scheduled and other background tasks.

 

Paul

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAKW6fif6cuHw%2BXJMbdO%3DML1czmoZLY-dScMw3Cm9od8sfVm1Hw%40mail.gmail.com.

[Stuart Douglas]

This is still a problem that we need to resolve, in that we have not yet merged our blocking executor (the main one provided by ExecutorRecorder), with the vert.x blocking task pool.

Almost everything uses our main one, however some vert.x stuff might use the vert.x blocking one. At the moment this is hard coded in Vert.x, so we can't really change it, so we need to discuss with the vert.x team how to go about this.

Stuart

To view this discussion on the web visit https://groups.google.com/d/msgid/quarkus-dev/CAKW6fif6cuHw%2BXJMbdO%3DML1czmoZLY-dScMw3Cm9od8sfVm1Hw%40mail.gmail.com.

[clement escoffier]

I agree. I need to check if it's still on the Vert.x roadmap. It may have been removed.

Clement