Right way to deal with Server Sent Events and don't block main event loop

[tchiotludo]

Hello,

I have some Server Sent Events events api endpoint.

Sometime time, I've some long operation on this enpoint and it seems that this operation will block the whole event loop from my api. Resulting all the endpoint from api and also healthcheck endpoint is blocked and will lead to a timeout.

I've the feeling I'm doing the right way on the controller (using and the @ExecuteOn(TaskExecutors.IO)) repository (using a Flowable.generate(()) but I've a lots of report that tell me it's not the case.

I'll try to reproduce the issue with some Thread.sleep around this method, but I'm not able to reproduce it.

Is this the right way to do SSE with micronaut ?

[graemerocher]

Do you have a way to reproduce the problem?

[tchiotludo]

No, I never be able to reproduce the issue.
I'll try a lot of stuff here, doing Thread.sleep(100000) inside the Flowable or outside and I never be able to reproduce the issue.
I've also another tools in production that I follow everyday and I noted the issues without a reproduction.
I know that the events loop is blocked because my GKE readiness failed with timeout.

FYI: this error come at least 50 times / week.

So I have the feeling that I doing this wrong, it's not the case ?

[graemerocher]

It is probably because the Netty event loop is by default 2 * number of processors and locally you have a lot more processors available than on GKE

A way to reproduce it could be running the application in a docker container limited to 1 or 2 CPUs

[tchiotludo]

Thanks @graemerocher !!

I'm able to reproduce the issue with a simple Thread.sleep on the Flowable.<>>create sending more http request than my cpu * 2.
The endpoint is timeout in this case.

In order to avoid the issue, I add a simple .observeOn(Schedulers.io()) :

        return Flowable
            .<Event<Execution>>create(emitter -> {
                // ....
            }, BackpressureStrategy.BUFFER)
            .observeOn(Schedulers.io())

That seems to prevent the issue and healthcheck continue to work even if there is more request than cpu * 2.

But since my Controller have @ExecuteOn(TaskExecutors.IO), is it not the goal of this annotation automatically observeOn on the TaskExecutors.IO ?

Is this a bug or documentation clarification that is needed ?

[graemerocher]

Seems like a bug, can you report an issue?

[tchiotludo]

done here : #5123 !
Thanks a lot :)