Emitter AMQP protocol stucked after MQ broker restart

[bkalas]

Describe the bug

we are using Emitters for sending message to remote MQ broker.
Normally without any OnOverflow annotation
@Inject @Channel(RESPONSE_QUEUE) Emitter<String> emitter; ... public void emit(Message<String> toEmit) { this.emitter.send(toEmit); ...
Few times we observed situation that if remote broker was restarted, emiting of messages stopped working.
For few x (~100) messages we didnt get any errror (i guess x= default size of buffer),
then we started to get errror
java.lang.IllegalStateException: SRMSG00034: Insufficient downstream requests to emit item
This was only solved by quarkus app restart.
Health report for channel was all time showing OK status.

We are also using a lot of consumers (@Incoming ) of amqp messages, this were susccessfully reconnected and continue to work.

Whn i try to use for example
@OnOverflow(FAIL, bufferSize = 100)
it at least discnnected after error from channel and started reporting KO in health check, but here bufferSize seems to be ignored, so it is not usable.

Expected behavior

Emitting channel must be resilient against remote broker restart and must continue to emit messages.

Actual behavior

Sometimes restarts of remote broker was handled ok, but noit always.

How to Reproduce?

1. must have installed amq broker (in our case Redhat AMQ7.11)
2. quarkus app with Emitter to some queue in remote broker
3. emit messages in some intervlas and restart broker.
4. for me always ~6 restart of broker was enough to simulate this issue, that aftger one of this restarts no new message ws emitted.

Output of uname -a or ver

No response

Output of java -version

17

Quarkus version or git rev

2.16.12.Final

Build tool (ie. output of mvnw --version or gradlew --version)

No response

Additional information

No response

[quarkus-bot]

/cc @cescoffier (reactive-messaging), @ozangunalp (reactive-messaging)

[ozangunalp]

Looks like an issue with the send-retry mechanism on the AMQP connector. It doesn't reconnect the client and keeps retrying with the previous (unconnected) sender.

I have a fix in mind but I need to be able to reproduce this scenario (in the test environment) to test it.

[MikkoKauhanen]

Hi,

If it helps at all I created quick and dirty test to try to reproduce the issue which you can find from link to the repo. This test uses Quarkus 3.10.1.

We have same problem in our services that use quarkus version 3.5.3. In our emitters we have @OnOverflow(OnOverflow.Strategy.UNBOUNDED_BUFFER) as a strategy.

Here is also link to my msg about my findings related to this issue: link to comment

[ozangunalp]

@MikkoKauhanen Thanks for this, I'll check it later today or at the beginning of next week.