[BUG] ReactorDispatcher instance is closed. Connection is lost

[the-mod]

Describe the bug
I am using azure-messaging-eventhubs:5.4 to span EventProcessingHost Instances on multiple Nodes and facing this Issue after the Services is running fine for several days.
My Apps then doesn't cosume Events anymore.
It seams something is off using the AccessToken (see second Exception), I also don't have Key Rotation applied on my Eventhubs.

Exception or Stack Trace

java.util.concurrent.RejectedExecutionException:
   at com.azure.core.amqp.implementation.ReactorDispatcher.throwIfSchedulerError (ReactorDispatcher.java93)
   at com.azure.core.amqp.implementation.ReactorDispatcher.invoke (ReactorDispatcher.java68)
   at com.azure.core.amqp.implementation.ReactorReceiver.dispose (ReactorReceiver.java190)
   at com.azure.core.amqp.implementation.ReactorSession$LinkSubscription.dispose (ReactorSession.java555)
   at com.azure.core.amqp.implementation.ReactorSession.lambda$dispose$1 (ReactorSession.java149)
   at java.util.concurrent.ConcurrentHashMap.forEach
   at com.azure.core.amqp.implementation.ReactorSession.dispose (ReactorSession.java149)
   at com.azure.core.amqp.implementation.ReactorConnection$SessionSubscription.dispose (ReactorConnection.java421)
   at com.azure.core.amqp.implementation.ReactorConnection.removeSession (ReactorConnection.java322)
   at com.azure.core.amqp.implementation.ReactorConnection.dispose (ReactorConnection.java266)
   at com.azure.core.amqp.implementation.ReactorConnection.dispose (ReactorConnection.java250)
   at com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection.dispose (EventHubReactorAmqpConnection.java144)
   at com.azure.core.amqp.implementation.AmqpChannelProcessor.close (AmqpChannelProcessor.java308)
   at com.azure.core.amqp.implementation.AmqpChannelProcessor.setAndClearChannel (AmqpChannelProcessor.java297)
   at com.azure.core.amqp.implementation.AmqpChannelProcessor.lambda$onNext$2 (AmqpChannelProcessor.java103)
   at reactor.core.publisher.LambdaSubscriber.onError (LambdaSubscriber.java149)
   at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal (FluxReplay.java802)
   at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replay (FluxReplay.java898)
   at reactor.core.publisher.ReplayProcessor.onError (ReplayProcessor.java440)
   at com.azure.core.amqp.implementation.ReactorConnection$ReactorExceptionHandler.onConnectionError (ReactorConnection.java380)
   at com.azure.core.amqp.implementation.ReactorExecutor.run (ReactorExecutor.java126)
   at reactor.core.scheduler.SchedulerTask.call (SchedulerTask.java68)
   at reactor.core.scheduler.SchedulerTask.call (SchedulerTask.java28)
   at java.util.concurrent.FutureTask.run
   at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run
   at java.util.concurrent.ThreadPoolExecutor.runWorker
   at java.util.concurrent.ThreadPoolExecutor$Worker.run
   at java.lang.Thread.run

The SDK itself logs (com.azure.core.util.logging.ClientLogger) this message (no Stacktrace available) comming from com.azure.core.amqp.implementation.ActiveClientTokenManager:

Error occurred while refreshing token that is not retriable. Not scheduling refresh task. Use ActiveClientTokenManager.authorize() to schedule task again. audience[amqp://host/eh/ConsumerGroups/cg/Partitions/9] scopes[amqp://host/eh/ConsumerGroups/cg/Partitions/9]
ReactorDispatcher instance is closed.

To Reproduce
Implement the EventProcessorHost and wait

Setup (please complete the following information):

• OS: Win10
• Version of the Library used: 5.4

[joshfree]

Thanks for filing this azure-messaging-eventhubs github issue, @the-mod. @srnagar / @conniey could you please follow up?

[conniey]

Related or this is a duplicate of #19753. We should appropriately close the link and recreate it though.

[the-mod]

@conniey thanks for the info. I updated today all my apps to azure-messaging-eventhubs:5.5.0 in order to get rid of this. But as it seams the Problems still exist

[the-mod]

@conniey Just to give you an update: its also happening with azure-messaging-eventhubs:5.5.0. Its also not only the EventProcessingHost I am using, it also happens with my EventHubProducerAsyncClient implementations.

It is giving me this error (w/o Stacktrace):

partitionId[null]: Sending messages timed out. ReactorDispatcher instance is closed.

coming from com.azure.core.amqp.implementation.RetryUtil

in combination with

ReactorDispatcher instance is closed.

coming from com.azure.core.amqp.implementation.ReactorDispatcher

The latest Error happens just now, before that on Saturday. So roughly 48h in between. Maybe it helps for the investigations

[the-mod]

@conniey @srnagar any news on this?

[conniey]

Hey,

I am currently looking at another reliability issue #18070 which may be related to your issue. But I will get to this issue afterwards because we should be creating a new link if there is an auth issue.

[the-mod]

@conniey thx for the update.

I just saw in my Logs, a partition was closed and opened again and again while erroring with this.
(Not sure if it is related to this Issue)

Exception: reactor.core.Exceptions$OverflowException

Message: The receiver is overrun by more signals than expected (bounded queue...)

Stacktrace:

reactor.core.Exceptions$OverflowException:
   at reactor.core.Exceptions.failWithOverflow (Exceptions.java221)
   at reactor.core.publisher.FluxWindowTimeout$WindowTimeoutSubscriber.onNext (FluxWindowTimeout.java233)
   at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext (FluxPeekFuseable.java203)
   at reactor.core.publisher.FluxDoFinally$DoFinallySubscriber.onNext (FluxDoFinally.java123)
   at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync (FluxPublishOn.java439)
   at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run (FluxPublishOn.java526)
   at reactor.core.scheduler.WorkerTask.call (WorkerTask.java84)
   at reactor.core.scheduler.WorkerTask.call (WorkerTask.java37)
   at java.util.concurrent.FutureTask.run
   at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run
   at java.util.concurrent.ThreadPoolExecutor.runWorker
   at java.util.concurrent.ThreadPoolExecutor$Worker.run
   at java.lang.Thread.run

[conniey]

@the-mod

We released 5.7.0 this morning. This should fix the unauthorized issue. It will close the send link on an unauthorized and try reauthorizing next time you try to send a message by creating another link.

https://repo1.maven.org/maven2/com/azure/azure-messaging-eventhubs/5.7.0/
https://repo1.maven.org/maven2/com/azure/azure-messaging-eventhubs-checkpointstore-blob/1.6.0/

Cheers,
Connie

[the-mod]

@conniey thank you for the Information. I will test it these days and provide Feedback.

[the-mod]

@conniey
I replaced in my project
azure-messaging-eventhubs: 5.6.0 and azure-messaging-eventhubs-checkpointstore-blob: 1.5.0
with
azure-messaging-eventhubs: 5.7.0 and azure-messaging-eventhubs-checkpointstore-blob: 1.6.0
and unfortunately I see a performance decrease by max 50%.

I also testet
azure-messaging-eventhubs: 5.6.0 and azure-messaging-eventhubs-checkpointstore-blob: 1.6.0
as well as
azure-messaging-eventhubs: 5.7.0 and azure-messaging-eventhubs-checkpointstore-blob: 1.5.0
with the same decreasing performance.

I didn't dive into this yet in detail.
Would be nice if you can check this behavior on your site.

Best

[conniey]

Fascinating. We did not do any performance tests on both solutions but there were several race conditions fixed, so I can see that affecting it. It may be best to create a new bug to address this.

/cc @YijunXieMS

[conniey]

Performance investigations #20791

[conniey]

Created an issue to track the performance degradation. #20841