Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[BUG] EventProcessorClient is running, but no events are received #13785

Closed
jonjarvis opened this issue Aug 4, 2020 · 2 comments · Fixed by #14990
Closed

[BUG] EventProcessorClient is running, but no events are received #13785

jonjarvis opened this issue Aug 4, 2020 · 2 comments · Fixed by #14990
Assignees
@srnagar
Labels
bug This issue requires a change to an existing behavior in the product in order to be resolved. Client This issue points to a problem in the data-plane of the library. customer-reported Issues that are reported by GitHub users external to the Azure organization. Event Hubs pillar-reliability The issue is related to reliability, one of our core engineering pillars. (includes stress testing)

Comments

@jonjarvis
Copy link

Describe the bug
I have a 2 partition event hub setup with a single consumer group. I start a single EventProcessorClient, and after some period of time ~30min (where I provide no events) the processor errors out in the logs. However, my process continues running, and doesn't know that the EventProcessorClient is no longer receiving events.

Exception or Stack Trace

020-08-04 14:00:23,809 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[testhub1]: Disposing of session.
2020-08-04 14:00:23,809 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkRemoteClose connectionId[MF_78cd35_1596565512594], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:141005cf6fe541efa3ae54458901e159_G27, SystemTracker:gateway5, Timestamp:2020-08-04T19:00:23]
2020-08-04 14:00:23,810 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] processOnClose connectionId[MF_78cd35_1596565512594], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:141005cf6fe541efa3ae54458901e159_G27, SystemTracker:gateway5, Timestamp:2020-08-04T19:00:23]
2020-08-04 14:00:23,810 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkRemoteClose connectionId[MF_78cd35_1596565512594], linkName[cbs:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:141005cf6fe541efa3ae54458901e159_G27, SystemTracker:gateway5, Timestamp:2020-08-04T19:00:23]
2020-08-04 14:00:23,810 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] processOnClose connectionId[MF_78cd35_1596565512594], linkName[cbs:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:141005cf6fe541efa3ae54458901e159_G27, SystemTracker:gateway5, Timestamp:2020-08-04T19:00:23]
2020-08-04 14:00:23,810 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose connectionId[cbs-session], entityName[MF_78cd35_1596565512594], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:00:23,810 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose closing a local session for connectionId[MF_78cd35_1596565512594], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:00:23,810 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: CLOSED
2020-08-04 14:00:23,810 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_78cd35_1596565512594] sessionName[cbs-session]: Complete. Removing and disposing session.
2020-08-04 14:00:23,811 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:00:23,811 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionRemoteClose hostname[############.servicebus.windows.net:5671], connectionId[MF_78cd35_1596565512594], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:141005cf6fe541efa3ae54458901e159_G27, SystemTracker:gateway5, Timestamp:2020-08-04T19:00:23]
2020-08-04 14:00:23,811 DEBUG [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_78cd35_1596565512594]: Connection state: CLOSED
2020-08-04 14:00:23,811 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Channel is closed.
2020-08-04 14:00:23,811 INFO  [com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection] connectionId[MF_78cd35_1596565512594]: Disposing of connection.
2020-08-04 14:00:23,811 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_78cd35_1596565512594]: Disposing of ReactorConnection.
2020-08-04 14:00:23,811 INFO  [com.azure.core.amqp.implementation.AmqpExceptionHandler] Shutdown received: ReactorExecutor.close() was called., isTransient[false], initiatedByClient[true]
2020-08-04 14:00:25,344 INFO  [com.azure.messaging.eventhubs.PartitionBasedLoadBalancer] Starting load balancer for d9ab75bd-3ae8-4a9f-b212-9418dc2a9f77
2020-08-04 14:00:25,533 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Found blob for partition ############.servicebus.windows.net/testhub1/$default/ownership/0
2020-08-04 14:00:25,533 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Blob ############.servicebus.windows.net/testhub1/$default/ownership/0 is owned by d9ab75bd-3ae8-4a9f-b212-9418dc2a9f77
2020-08-04 14:00:25,533 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Found blob for partition ############.servicebus.windows.net/testhub1/$default/ownership/1
2020-08-04 14:00:25,534 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Blob ############.servicebus.windows.net/testhub1/$default/ownership/1 is owned by d9ab75bd-3ae8-4a9f-b212-9418dc2a9f77
2020-08-04 14:00:28,323 ERROR [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_78cd35_1596565512594]: Connection is disposed. Cannot get CBS node.
java.lang.IllegalStateException: connectionId[MF_78cd35_1596565512594]: Connection is disposed. Cannot get CBS node.
	at com.azure.core.amqp.implementation.ReactorConnection.getClaimsBasedSecurityNode(ReactorConnection.java:147)
	at com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection.createSession(EventHubReactorAmqpConnection.java:150)
	at com.azure.core.amqp.implementation.ReactorConnection.lambda$createSession$9(ReactorConnection.java:212)
	at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1660)
	at com.azure.core.amqp.implementation.ReactorConnection.lambda$createSession$10(ReactorConnection.java:206)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
	at reactor.core.publisher.Operators$MonoSubscriber.request(Operators.java:1842)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2089)
	at com.azure.core.amqp.implementation.AmqpChannelProcessor.requestUpstream(AmqpChannelProcessor.java:286)
	at com.azure.core.amqp.implementation.AmqpChannelProcessor.lambda$onError$4(AmqpChannelProcessor.java:172)
	at reactor.core.publisher.LambdaMonoSubscriber.onNext(LambdaMonoSubscriber.java:168)
	at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:117)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2020-08-04 14:00:28,324 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: UNINITIALIZED
2020-08-04 14:00:28,324 INFO  [com.azure.core.amqp.implementation.ReactorConnection] Emitting new response channel. connectionId: MF_78cd35_1596565512594. entityPath: $cbs. linkName: cbs.
2020-08-04 14:00:35,348 INFO  [com.azure.messaging.eventhubs.PartitionBasedLoadBalancer] Load balancer already running
2020-08-04 14:00:45,350 INFO  [com.azure.messaging.eventhubs.PartitionBasedLoadBalancer] Load balancer already running
2020-08-04 14:00:55,352 INFO  [com.azure.messaging.eventhubs.PartitionBasedLoadBalancer] Load balancer already running
2020-08-04 14:01:05,353 INFO  [com.azure.messaging.eventhubs.PartitionBasedLoadBalancer] Load balancer already running
2020-08-04 14:01:15,355 INFO  [com.azure.messaging.eventhubs.PartitionBasedLoadBalancer] Load balancer already running
2020-08-04 14:01:23,814 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] Unable to acquire dispose reactor semaphore within timeout.
2020-08-04 14:01:23,814 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[testhub1], entityName[MF_78cd35_1596565512594], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:01:23,815 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_78cd35_1596565512594], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:141005cf6fe541efa3ae54458901e159_G27, SystemTracker:gateway5, Timestamp:2020-08-04T19:00:23]
2020-08-04 14:01:23,815 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_78cd35_1596565512594], linkName[cbs:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:01:23,815 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_78cd35_1596565512594], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:01:23,815 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionLocalClose hostname[############.servicebus.windows.net:5671], connectionId[MF_78cd35_1596565512594], errorCondition[null], errorDescription[null]
2020-08-04 14:01:23,816 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalOpen connectionId[MF_78cd35_1596565512594], entityName[cbs-session], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:01:23,816 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionUnbound hostname[############.servicebus.windows.net:5671], connectionId[MF_78cd35_1596565512594], state[CLOSED], remoteState[CLOSED]
2020-08-04 14:01:23,816 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_78cd35_1596565512594],  linkName[testhub1]
2020-08-04 14:01:23,816 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_78cd35_1596565512594], entityName[testhub1], condition[null], description[null]
2020-08-04 14:01:23,816 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_78cd35_1596565512594],  linkName[cbs:sender]
2020-08-04 14:01:23,817 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_78cd35_1596565512594],  linkName[cbs:receiver]
2020-08-04 14:01:23,817 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_78cd35_1596565512594], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:01:23,817 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_78cd35_1596565512594],  linkName[cbs:sender]
2020-08-04 14:01:23,817 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_78cd35_1596565512594],  linkName[cbs:receiver]
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_78cd35_1596565512594], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_78cd35_1596565512594] sessionName[cbs-session]: Complete. Removing and disposing session.
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionFinal hostname[############.servicebus.windows.net:5671], connectionId[MF_78cd35_1596565512594], errorCondition[null], errorDescription[null]
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_78cd35_1596565512594], linkName[cbs:sender], errorCondition[null], errorDescription[null]
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_78cd35_1596565512594], linkName[cbs:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:01:23,819 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_78cd35_1596565512594], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:01:23,819 DEBUG [com.azure.core.amqp.implementation.handler.DispatchHandler] Running task for event: %s
2020-08-04 14:01:23,819 DEBUG [com.azure.core.amqp.implementation.handler.DispatchHandler] Running task for event: %s
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_78cd35_1596565512594], message[Processing all pending tasks and closing old reactor.]
2020-08-04 14:01:23,819 DEBUG [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalOpen connectionId[MF_78cd35_1596565512594], linkName[cbs:sender], localTarget[Target{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}]
2020-08-04 14:01:23,819 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalOpen connectionId[MF_78cd35_1596565512594], linkName[cbs:receiver], localSource[Source{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=null, capabilities=null}]
2020-08-04 14:01:23,820 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_78cd35_1596565512594], message[Stopping the reactor because thread was interrupted or the reactor has no more events to process.]
2020-08-04 14:01:23,820 DEBUG [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFlow connectionId[MF_83fe8b_1596565403336], linkName[mgmt:sender], unsettled[0], credit[81]
2020-08-04 14:01:23,863 WARN  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onTransportError hostname[############.servicebus.windows.net:5671], connectionId[MF_e1bd31_1596565512592], error[An existing connection was forcibly closed by the remote host]
2020-08-04 14:01:23,863 ERROR [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_e1bd31_1596565512592] Error occurred in connection handler.
An existing connection was forcibly closed by the remote host, errorContext[NAMESPACE: ############.servicebus.windows.net]
com.azure.core.amqp.exception.AmqpException: An existing connection was forcibly closed by the remote host, errorContext[NAMESPACE: ############.servicebus.windows.net]
	at com.azure.core.amqp.implementation.ExceptionUtil.toException(ExceptionUtil.java:78)
	at com.azure.core.amqp.implementation.handler.ConnectionHandler.notifyErrorContext(ConnectionHandler.java:262)
	at com.azure.core.amqp.implementation.handler.ConnectionHandler.onTransportError(ConnectionHandler.java:154)
	at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:191)
	at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
	at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:82)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2020-08-04 14:01:23,863 INFO  [com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection] connectionId[MF_e1bd31_1596565512592]: Disposing of connection.
2020-08-04 14:01:23,863 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_e1bd31_1596565512592]: Disposing of ReactorConnection.
2020-08-04 14:01:23,863 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_e1bd31_1596565512592]: Removing session 'testhub1'
2020-08-04 14:01:23,864 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[testhub1]: Disposing of session.
2020-08-04 14:01:23,864 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_e1bd31_1596565512592]: Removing session 'cbs-session'
2020-08-04 14:01:23,864 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:01:23,864 INFO  [com.azure.core.amqp.implementation.AmqpExceptionHandler] Shutdown received: ReactorExecutor.close() was called., isTransient[false], initiatedByClient[true]
2020-08-04 14:01:25,347 WARN  [com.azure.messaging.eventhubs.PartitionBasedLoadBalancer] Load balancing for event processor failed - Did not observe any item or terminal signal within 60000ms in 'flatMapMany' (and no fallback has been configured)
Did not observe any item or terminal signal within 60000ms in 'flatMapMany' (and no fallback has been configured)
java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 60000ms in 'flatMapMany' (and no fallback has been configured)
	at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.handleTimeout(FluxTimeout.java:289)
	at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.doTimeout(FluxTimeout.java:274)
	at reactor.core.publisher.FluxTimeout$TimeoutTimeoutSubscriber.onNext(FluxTimeout.java:396)
	at reactor.core.publisher.StrictSubscriber.onNext(StrictSubscriber.java:89)
	at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73)
	at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:117)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2020-08-04 14:01:25,349 INFO  [com.azure.core.amqp.implementation.ReactorSession] linkName[0_359d57_1596565405425] entityPath[testhub1/ConsumerGroups/$default/Partitions/0]: Complete. Removing receive link.
2020-08-04 14:01:25,349 INFO  [com.azure.messaging.eventhubs.EventHubConsumerAsyncClient] linkName[0_359d57_1596565405425], partitionId[0], signal[onComplete]: Receiving completed.
2020-08-04 14:01:25,349 INFO  [com.azure.messaging.eventhubs.implementation.PartitionProcessor] Closing partition processor for partition 0 with close reason EVENT_PROCESSOR_SHUTDOWN
2020-08-04 14:01:25,350 INFO  [com.azure.messaging.eventhubs.implementation.AmqpReceiveLinkProcessor] Terminal state reached. Disposing of link processor.
2020-08-04 14:01:25,350 INFO  [com.azure.messaging.eventhubs.PartitionPumpManager] Closing consumer for partition id 0
2020-08-04 14:01:25,351 INFO  [com.azure.messaging.eventhubs.PartitionPumpManager] Removing partition id 0 from list of processing partitions
2020-08-04 14:01:25,352 INFO  [com.azure.messaging.eventhubs.implementation.AmqpReceiveLinkProcessor] Disposing receive link link name: [0_359d57_1596565405425], entity path: [testhub1/ConsumerGroups/$default/Partitions/0]
2020-08-04 14:01:25,354 INFO  [com.azure.messaging.eventhubs.EventHubPartitionAsyncConsumer] Closed consumer for partition 0
2020-08-04 14:01:25,354 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] Upstream connection publisher was completed. Terminating processor.
2020-08-04 14:01:25,355 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: AMQP channel processor completed. Notifying 0 subscribers.
2020-08-04 14:01:25,355 INFO  [com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection] connectionId[MF_112726_1596565405424]: Disposing of connection.
2020-08-04 14:01:25,355 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_112726_1596565405424]: Disposing of ReactorConnection.
2020-08-04 14:01:25,356 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Channel is disposed.
2020-08-04 14:01:25,356 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_112726_1596565405424]: Removing session 'testhub1/ConsumerGroups/$default/Partitions/0'
2020-08-04 14:01:25,356 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[testhub1/ConsumerGroups/$default/Partitions/0]: Disposing of session.
2020-08-04 14:01:25,357 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_112726_1596565405424]: Removing session 'cbs-session'
2020-08-04 14:01:25,357 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:01:25,357 INFO  [com.azure.core.amqp.implementation.AmqpExceptionHandler] Shutdown received: ReactorExecutor.close() was called., isTransient[false], initiatedByClient[true]
2020-08-04 14:02:23,866 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] Unable to acquire dispose reactor semaphore within timeout.
2020-08-04 14:02:23,866 DEBUG [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] Attempted 1 times to get a new AMQP connection
2020-08-04 14:02:23,866 WARN  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] Retry #1. Transient error occurred. Retrying after 4511 ms.
An existing connection was forcibly closed by the remote host, errorContext[NAMESPACE: ############.servicebus.windows.net]
com.azure.core.amqp.exception.AmqpException: An existing connection was forcibly closed by the remote host, errorContext[NAMESPACE: ############.servicebus.windows.net]
	at com.azure.core.amqp.implementation.ExceptionUtil.toException(ExceptionUtil.java:78)
	at com.azure.core.amqp.implementation.handler.ConnectionHandler.notifyErrorContext(ConnectionHandler.java:262)
	at com.azure.core.amqp.implementation.handler.ConnectionHandler.onTransportError(ConnectionHandler.java:154)
	at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:191)
	at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
	at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:82)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2020-08-04 14:02:23,866 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[testhub1], entityName[MF_e1bd31_1596565512592], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:02:23,866 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_e1bd31_1596565512592], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:02:23,866 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionLocalClose hostname[############.servicebus.windows.net:5671], connectionId[MF_e1bd31_1596565512592], errorCondition[null], errorDescription[null]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionUnbound hostname[############.servicebus.windows.net:5671], connectionId[MF_e1bd31_1596565512592], state[CLOSED], remoteState[ACTIVE]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_e1bd31_1596565512592],  linkName[testhub1]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_e1bd31_1596565512592], entityName[testhub1], condition[null], description[null]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_e1bd31_1596565512592],  linkName[cbs:sender]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_e1bd31_1596565512592],  linkName[cbs:receiver]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_e1bd31_1596565512592], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionFinal hostname[############.servicebus.windows.net:5671], connectionId[MF_e1bd31_1596565512592], errorCondition[null], errorDescription[null]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_e1bd31_1596565512592], linkName[cbs:sender], errorCondition[null], errorDescription[null]
2020-08-04 14:02:23,867 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_e1bd31_1596565512592], linkName[cbs:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:02:23,868 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_112726_1596565405424], message[Stopping the reactor because thread was interrupted or the reactor has no more events to process.]
2020-08-04 14:02:23,868 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_e1bd31_1596565512592], message[Processing all pending tasks and closing old reactor.]
2020-08-04 14:02:23,869 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_112726_1596565405424], message[Processing all pending tasks and closing old reactor.]
2020-08-04 14:02:23,870 DEBUG [com.azure.core.amqp.implementation.handler.DispatchHandler] Running task for event: %s
2020-08-04 14:02:23,870 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_e1bd31_1596565512592], message[Stopping the reactor because thread was interrupted or the reactor has no more events to process.]
2020-08-04 14:02:23,870 INFO  [com.azure.core.amqp.implementation.ReactorSession] linkName[1_1eb6a3_1596565415136] entityPath[testhub1/ConsumerGroups/$default/Partitions/1]: Complete. Removing receive link.
2020-08-04 14:02:23,870 INFO  [com.azure.messaging.eventhubs.EventHubConsumerAsyncClient] linkName[1_1eb6a3_1596565415136], partitionId[1], signal[onComplete]: Receiving completed.
2020-08-04 14:02:23,870 INFO  [com.azure.messaging.eventhubs.implementation.PartitionProcessor] Closing partition processor for partition 1 with close reason EVENT_PROCESSOR_SHUTDOWN
2020-08-04 14:02:23,870 DEBUG [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onDelivery connectionId[MF_83fe8b_1596565403336], linkName[mgmt:receiver], updatedLinkCredit[0], remoteCredit[0], remoteCondition[Error{condition=null, description='null', info=null}], delivery.isPartial[false]
2020-08-04 14:02:23,870 INFO  [com.azure.messaging.eventhubs.implementation.AmqpReceiveLinkProcessor] Terminal state reached. Disposing of link processor.
2020-08-04 14:02:23,871 INFO  [com.azure.messaging.eventhubs.implementation.AmqpReceiveLinkProcessor] Disposing receive link link name: [1_1eb6a3_1596565415136], entity path: [testhub1/ConsumerGroups/$default/Partitions/1]
2020-08-04 14:02:23,871 INFO  [com.azure.messaging.eventhubs.PartitionPumpManager] Closing consumer for partition id 1
2020-08-04 14:02:23,871 INFO  [com.azure.messaging.eventhubs.EventHubPartitionAsyncConsumer] Closed consumer for partition 1
2020-08-04 14:02:23,871 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] Upstream connection publisher was completed. Terminating processor.
2020-08-04 14:02:23,872 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: AMQP channel processor completed. Notifying 0 subscribers.
2020-08-04 14:02:23,872 INFO  [com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection] connectionId[MF_705c60_1596565415135]: Disposing of connection.
2020-08-04 14:02:23,872 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_705c60_1596565415135]: Disposing of ReactorConnection.
2020-08-04 14:02:23,872 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Channel is disposed.
2020-08-04 14:02:23,872 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_705c60_1596565415135]: Removing session 'testhub1/ConsumerGroups/$default/Partitions/1'
2020-08-04 14:02:23,872 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[testhub1/ConsumerGroups/$default/Partitions/1]: Disposing of session.
2020-08-04 14:02:23,872 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_705c60_1596565415135]: Removing session 'cbs-session'
2020-08-04 14:02:23,872 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:02:23,872 INFO  [com.azure.core.amqp.implementation.AmqpExceptionHandler] Shutdown received: ReactorExecutor.close() was called., isTransient[false], initiatedByClient[true]
2020-08-04 14:02:23,872 INFO  [com.azure.messaging.eventhubs.PartitionPumpManager] Removing partition id 1 from list of processing partitions
2020-08-04 14:02:23,932 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_705c60_1596565415135], message[Stopping the reactor because thread was interrupted or the reactor has no more events to process.]
2020-08-04 14:02:23,932 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_705c60_1596565415135], message[Processing all pending tasks and closing old reactor.]
2020-08-04 14:02:23,932 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_705c60_1596565415135], linkName[1_1eb6a3_1596565415136], errorCondition[null], errorDescription[null]
2020-08-04 14:02:23,933 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[testhub1/ConsumerGroups/$default/Partitions/1], entityName[MF_705c60_1596565415135], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:02:23,933 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_705c60_1596565415135], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:02:23,933 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionLocalClose hostname[############.servicebus.windows.net:5671], connectionId[MF_705c60_1596565415135], errorCondition[null], errorDescription[null]
2020-08-04 14:02:23,935 DEBUG [com.azure.core.amqp.implementation.handler.DispatchHandler] Running task for event: %s
2020-08-04 14:02:24,118 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Found blob for partition ############.servicebus.windows.net/testhub1/$default/ownership/0
2020-08-04 14:02:24,118 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Blob ############.servicebus.windows.net/testhub1/$default/ownership/0 is owned by d9ab75bd-3ae8-4a9f-b212-9418dc2a9f77
2020-08-04 14:02:24,119 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Found blob for partition ############.servicebus.windows.net/testhub1/$default/ownership/1
2020-08-04 14:02:24,119 DEBUG [com.azure.messaging.eventhubs.checkpointstore.blob.BlobCheckpointStore] Blob ############.servicebus.windows.net/testhub1/$default/ownership/1 is owned by d9ab75bd-3ae8-4a9f-b212-9418dc2a9f77
2020-08-04 14:02:28,378 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] Retry #1. Requesting from upstream.
2020-08-04 14:02:28,378 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Connection not requested, yet. Requesting one.
2020-08-04 14:02:28,378 INFO  [com.azure.messaging.eventhubs.EventHubClientBuilder] connectionId[MF_2f3162_1596567748378]: Emitting a single connection.
2020-08-04 14:02:28,378 DEBUG [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_2f3162_1596567748378]: Connection state: UNINITIALIZED
2020-08-04 14:02:28,378 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Setting next AMQP channel.
2020-08-04 14:02:28,378 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Next AMQP channel received, updating 0 current subscribers
2020-08-04 14:04:06,163 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose connectionId[testhub1], entityName[MF_45a617_1596565734315], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:04:06,163 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose closing a local session for connectionId[MF_45a617_1596565734315], entityName[testhub1], condition[null], description[null]
2020-08-04 14:04:06,163 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: CLOSED
2020-08-04 14:04:06,163 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_45a617_1596565734315] sessionName[testhub1]: Complete. Removing and disposing session.
2020-08-04 14:04:06,163 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[testhub1]: Disposing of session.
2020-08-04 14:04:06,163 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[testhub1], entityName[MF_45a617_1596565734315], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:04:06,164 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkRemoteClose connectionId[MF_45a617_1596565734315], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:90c30e9da6304d5e8b8653d351e3e68c_G29, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:04:06,164 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] processOnClose connectionId[MF_45a617_1596565734315], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:90c30e9da6304d5e8b8653d351e3e68c_G29, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:04:06,164 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkRemoteClose connectionId[MF_45a617_1596565734315], linkName[cbs:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:90c30e9da6304d5e8b8653d351e3e68c_G29, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:04:06,165 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] processOnClose connectionId[MF_45a617_1596565734315], linkName[cbs:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:90c30e9da6304d5e8b8653d351e3e68c_G29, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:04:06,165 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose connectionId[cbs-session], entityName[MF_45a617_1596565734315], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:04:06,165 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose closing a local session for connectionId[MF_45a617_1596565734315], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:04:06,165 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: CLOSED
2020-08-04 14:04:06,165 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_45a617_1596565734315] sessionName[cbs-session]: Complete. Removing and disposing session.
2020-08-04 14:04:06,165 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:04:06,165 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionRemoteClose hostname[############.servicebus.windows.net:5671], connectionId[MF_45a617_1596565734315], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:90c30e9da6304d5e8b8653d351e3e68c_G29, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:04:06,165 DEBUG [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_45a617_1596565734315]: Connection state: CLOSED
2020-08-04 14:04:06,166 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Channel is closed.
2020-08-04 14:04:06,166 INFO  [com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection] connectionId[MF_45a617_1596565734315]: Disposing of connection.
2020-08-04 14:04:06,166 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_45a617_1596565734315]: Disposing of ReactorConnection.
2020-08-04 14:04:06,166 INFO  [com.azure.core.amqp.implementation.AmqpExceptionHandler] Shutdown received: ReactorExecutor.close() was called., isTransient[false], initiatedByClient[true]
2020-08-04 14:04:10,677 ERROR [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_45a617_1596565734315]: Connection is disposed. Cannot get CBS node.
java.lang.IllegalStateException: connectionId[MF_45a617_1596565734315]: Connection is disposed. Cannot get CBS node.
	at com.azure.core.amqp.implementation.ReactorConnection.getClaimsBasedSecurityNode(ReactorConnection.java:147)
	at com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection.createSession(EventHubReactorAmqpConnection.java:150)
	at com.azure.core.amqp.implementation.ReactorConnection.lambda$createSession$9(ReactorConnection.java:212)
	at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1660)
	at com.azure.core.amqp.implementation.ReactorConnection.lambda$createSession$10(ReactorConnection.java:206)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
	at reactor.core.publisher.Operators$MonoSubscriber.request(Operators.java:1842)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2089)
	at com.azure.core.amqp.implementation.AmqpChannelProcessor.requestUpstream(AmqpChannelProcessor.java:286)
	at com.azure.core.amqp.implementation.AmqpChannelProcessor.lambda$onError$4(AmqpChannelProcessor.java:172)
	at reactor.core.publisher.LambdaMonoSubscriber.onNext(LambdaMonoSubscriber.java:168)
	at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:117)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2020-08-04 14:04:10,677 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: UNINITIALIZED
2020-08-04 14:04:10,678 INFO  [com.azure.core.amqp.implementation.ReactorConnection] Emitting new response channel. connectionId: MF_45a617_1596565734315. entityPath: $cbs. linkName: cbs.
2020-08-04 14:05:06,168 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] Unable to acquire dispose reactor semaphore within timeout.
2020-08-04 14:05:06,168 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onTransportClosed hostname[############.servicebus.windows.net:5671], connectionId[MF_45a617_1596565734315], error[n/a]
2020-08-04 14:05:06,168 INFO  [com.azure.core.amqp.implementation.handler.CustomIOHandler] onTransportClosed connectionId[MF_45a617_1596565734315], hostname[############.servicebus.windows.net:5671]
2020-08-04 14:05:06,169 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_45a617_1596565734315], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:90c30e9da6304d5e8b8653d351e3e68c_G29, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:05:06,169 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_45a617_1596565734315], linkName[cbs:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:05:06,169 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_45a617_1596565734315], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:05:06,169 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionLocalClose hostname[############.servicebus.windows.net:5671], connectionId[MF_45a617_1596565734315], errorCondition[null], errorDescription[null]
2020-08-04 14:05:06,169 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalOpen connectionId[MF_45a617_1596565734315], entityName[cbs-session], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:05:06,170 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionUnbound hostname[############.servicebus.windows.net:5671], connectionId[MF_45a617_1596565734315], state[CLOSED], remoteState[CLOSED]
2020-08-04 14:05:06,170 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_45a617_1596565734315],  linkName[testhub1]
2020-08-04 14:05:06,170 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_45a617_1596565734315], entityName[testhub1], condition[null], description[null]
2020-08-04 14:05:06,170 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_45a617_1596565734315],  linkName[cbs:sender]
2020-08-04 14:05:06,170 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_45a617_1596565734315],  linkName[cbs:receiver]
2020-08-04 14:05:06,171 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_45a617_1596565734315], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:05:06,171 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_45a617_1596565734315],  linkName[cbs:sender]
2020-08-04 14:05:06,171 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_45a617_1596565734315],  linkName[cbs:receiver]
2020-08-04 14:05:06,171 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_45a617_1596565734315], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:05:06,171 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_45a617_1596565734315] sessionName[cbs-session]: Complete. Removing and disposing session.
2020-08-04 14:05:06,172 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:05:06,172 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionFinal hostname[############.servicebus.windows.net:5671], connectionId[MF_45a617_1596565734315], errorCondition[null], errorDescription[null]
2020-08-04 14:05:06,172 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_45a617_1596565734315], linkName[cbs:sender], errorCondition[null], errorDescription[null]
2020-08-04 14:05:06,172 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_45a617_1596565734315], linkName[cbs:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:05:06,172 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_45a617_1596565734315], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:05:06,173 DEBUG [com.azure.core.amqp.implementation.handler.DispatchHandler] Running task for event: %s
2020-08-04 14:05:06,194 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose connectionId[testhub1], entityName[MF_5599ad_1596565734312], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:05:06,194 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose closing a local session for connectionId[MF_5599ad_1596565734312], entityName[testhub1], condition[null], description[null]
2020-08-04 14:05:06,194 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: CLOSED
2020-08-04 14:05:06,194 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_5599ad_1596565734312] sessionName[testhub1]: Complete. Removing and disposing session.
2020-08-04 14:05:06,195 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[testhub1]: Disposing of session.
2020-08-04 14:05:06,195 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkRemoteClose connectionId[MF_5599ad_1596565734312], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:274fa2e014d340eeb364724cb125f31c_G8, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:05:06,195 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] processOnClose connectionId[MF_5599ad_1596565734312], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:274fa2e014d340eeb364724cb125f31c_G8, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:05:06,196 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkRemoteClose connectionId[MF_5599ad_1596565734312], linkName[cbs:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:274fa2e014d340eeb364724cb125f31c_G8, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:05:06,196 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] processOnClose connectionId[MF_5599ad_1596565734312], linkName[cbs:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:274fa2e014d340eeb364724cb125f31c_G8, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:05:06,196 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose connectionId[cbs-session], entityName[MF_5599ad_1596565734312], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:05:06,196 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose closing a local session for connectionId[MF_5599ad_1596565734312], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:05:06,196 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: CLOSED
2020-08-04 14:05:06,197 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_5599ad_1596565734312] sessionName[cbs-session]: Complete. Removing and disposing session.
2020-08-04 14:05:06,197 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:05:06,197 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionRemoteClose hostname[############.servicebus.windows.net:5671], connectionId[MF_5599ad_1596565734312], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:274fa2e014d340eeb364724cb125f31c_G8, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:05:06,197 DEBUG [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_5599ad_1596565734312]: Connection state: CLOSED
2020-08-04 14:05:06,197 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Channel is closed.
2020-08-04 14:05:06,198 INFO  [com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection] connectionId[MF_5599ad_1596565734312]: Disposing of connection.
2020-08-04 14:05:06,198 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_5599ad_1596565734312]: Disposing of ReactorConnection.
2020-08-04 14:05:06,198 INFO  [com.azure.core.amqp.implementation.AmqpExceptionHandler] Shutdown received: ReactorExecutor.close() was called., isTransient[false], initiatedByClient[true]
2020-08-04 14:05:10,707 ERROR [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_5599ad_1596565734312]: Connection is disposed. Cannot get CBS node.
java.lang.IllegalStateException: connectionId[MF_5599ad_1596565734312]: Connection is disposed. Cannot get CBS node.
	at com.azure.core.amqp.implementation.ReactorConnection.getClaimsBasedSecurityNode(ReactorConnection.java:147)
	at com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection.createSession(EventHubReactorAmqpConnection.java:150)
	at com.azure.core.amqp.implementation.ReactorConnection.lambda$createSession$9(ReactorConnection.java:212)
	at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1660)
	at com.azure.core.amqp.implementation.ReactorConnection.lambda$createSession$10(ReactorConnection.java:206)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.onNext(FluxMapFuseable.java:107)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.onNext(FluxPeekFuseable.java:203)
	at reactor.core.publisher.Operators$MonoSubscriber.request(Operators.java:1842)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxMapFuseable$MapFuseableSubscriber.request(FluxMapFuseable.java:162)
	at reactor.core.publisher.FluxPeekFuseable$PeekFuseableSubscriber.request(FluxPeekFuseable.java:137)
	at reactor.core.publisher.Operators$MultiSubscriptionSubscriber.request(Operators.java:2089)
	at com.azure.core.amqp.implementation.AmqpChannelProcessor.requestUpstream(AmqpChannelProcessor.java:286)
	at com.azure.core.amqp.implementation.AmqpChannelProcessor.lambda$onError$4(AmqpChannelProcessor.java:172)
	at reactor.core.publisher.LambdaMonoSubscriber.onNext(LambdaMonoSubscriber.java:168)
	at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:117)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2020-08-04 14:05:10,709 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: UNINITIALIZED
2020-08-04 14:05:10,711 INFO  [com.azure.core.amqp.implementation.ReactorConnection] Emitting new response channel. connectionId: MF_5599ad_1596565734312. entityPath: $cbs. linkName: cbs.
2020-08-04 14:06:06,200 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] Unable to acquire dispose reactor semaphore within timeout.
2020-08-04 14:06:06,200 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[testhub1], entityName[MF_5599ad_1596565734312], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:06:06,200 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_5599ad_1596565734312], linkName[cbs:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:274fa2e014d340eeb364724cb125f31c_G8, SystemTracker:gateway5, Timestamp:2020-08-04T19:04:06]
2020-08-04 14:06:06,200 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_5599ad_1596565734312], linkName[cbs:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:06:06,200 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_5599ad_1596565734312], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:06:06,200 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionLocalClose hostname[############.servicebus.windows.net:5671], connectionId[MF_5599ad_1596565734312], errorCondition[null], errorDescription[null]
2020-08-04 14:06:06,200 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalOpen connectionId[MF_5599ad_1596565734312], entityName[cbs-session], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:06:06,201 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionUnbound hostname[############.servicebus.windows.net:5671], connectionId[MF_5599ad_1596565734312], state[CLOSED], remoteState[CLOSED]
2020-08-04 14:06:06,201 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_5599ad_1596565734312],  linkName[testhub1]
2020-08-04 14:06:06,201 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_5599ad_1596565734312], entityName[testhub1], condition[null], description[null]
2020-08-04 14:06:06,201 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_5599ad_1596565734312],  linkName[cbs:sender]
2020-08-04 14:06:06,201 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_5599ad_1596565734312],  linkName[cbs:receiver]
2020-08-04 14:06:06,202 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_5599ad_1596565734312], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:06:06,202 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_5599ad_1596565734312],  linkName[cbs:sender]
2020-08-04 14:06:06,202 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_5599ad_1596565734312],  linkName[cbs:receiver]
2020-08-04 14:06:06,202 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_5599ad_1596565734312], entityName[cbs-session], condition[null], description[null]
2020-08-04 14:06:06,202 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_5599ad_1596565734312] sessionName[cbs-session]: Complete. Removing and disposing session.
2020-08-04 14:06:06,203 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[cbs-session]: Disposing of session.
2020-08-04 14:06:06,203 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionFinal hostname[############.servicebus.windows.net:5671], connectionId[MF_5599ad_1596565734312], errorCondition[null], errorDescription[null]
2020-08-04 14:06:06,203 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_5599ad_1596565734312], linkName[cbs:sender], errorCondition[null], errorDescription[null]
2020-08-04 14:06:06,203 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_5599ad_1596565734312], linkName[cbs:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:06:06,203 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[cbs-session], entityName[MF_5599ad_1596565734312], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:06:06,204 DEBUG [com.azure.core.amqp.implementation.handler.DispatchHandler] Running task for event: %s
2020-08-04 14:06:06,204 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_45a617_1596565734315], message[Processing all pending tasks and closing old reactor.]
2020-08-04 14:06:06,204 DEBUG [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalOpen connectionId[MF_45a617_1596565734315], linkName[cbs:sender], localTarget[Target{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}]
2020-08-04 14:06:06,204 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalOpen connectionId[MF_45a617_1596565734315], linkName[cbs:receiver], localSource[Source{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=null, capabilities=null}]
2020-08-04 14:06:06,205 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_45a617_1596565734315], message[Stopping the reactor because thread was interrupted or the reactor has no more events to process.]
2020-08-04 14:06:06,205 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_5599ad_1596565734312], message[Processing all pending tasks and closing old reactor.]
2020-08-04 14:06:06,205 DEBUG [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalOpen connectionId[MF_5599ad_1596565734312], linkName[cbs:sender], localTarget[Target{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, capabilities=null}]
2020-08-04 14:06:06,205 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalOpen connectionId[MF_5599ad_1596565734312], linkName[cbs:receiver], localSource[Source{address='$cbs', durable=NONE, expiryPolicy=SESSION_END, timeout=0, dynamic=false, dynamicNodeProperties=null, distributionMode=null, filter=null, defaultOutcome=null, outcomes=null, capabilities=null}]
2020-08-04 14:06:06,206 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_5599ad_1596565734312], message[Stopping the reactor because thread was interrupted or the reactor has no more events to process.]
2020-08-04 14:06:23,878 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkRemoteClose connectionId[MF_83fe8b_1596565403336], linkName[mgmt:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23]
2020-08-04 14:06:23,878 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] processOnClose connectionId[MF_83fe8b_1596565403336], linkName[mgmt:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23]
2020-08-04 14:06:23,879 ERROR [com.azure.messaging.eventhubs.implementation.ManagementChannel] Exception occurred:
The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23, errorContext[NAMESPACE: ############.servicebus.windows.net, PATH: $management, REFERENCE_ID: mgmt:sender, LINK_CREDIT: 81]
com.azure.core.amqp.exception.AmqpException: The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23, errorContext[NAMESPACE: ############.servicebus.windows.net, PATH: $management, REFERENCE_ID: mgmt:sender, LINK_CREDIT: 81]
	at com.azure.core.amqp.implementation.ExceptionUtil.toException(ExceptionUtil.java:78)
	at com.azure.core.amqp.implementation.handler.LinkHandler.processOnClose(LinkHandler.java:90)
	at com.azure.core.amqp.implementation.handler.LinkHandler.handleRemoteLinkClosed(LinkHandler.java:108)
	at com.azure.core.amqp.implementation.handler.LinkHandler.onLinkRemoteClose(LinkHandler.java:50)
	at com.azure.core.amqp.implementation.handler.SendLinkHandler.onLinkRemoteClose(SendLinkHandler.java:20)
	at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:176)
	at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
	at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
	at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:82)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
	at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
2020-08-04 14:06:23,880 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkLocalClose connectionId[MF_83fe8b_1596565403336], linkName[mgmt:sender], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23]
2020-08-04 14:06:23,881 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkLocalClose connectionId[MF_83fe8b_1596565403336], linkName[mgmt:receiver], errorCondition[null], errorDescription[null]
2020-08-04 14:06:23,883 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkRemoteClose connectionId[MF_83fe8b_1596565403336], linkName[mgmt:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23]
2020-08-04 14:06:23,883 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] processOnClose connectionId[MF_83fe8b_1596565403336], linkName[mgmt:receiver], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23]
2020-08-04 14:06:23,883 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose connectionId[mgmt-session], entityName[MF_83fe8b_1596565403336], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:06:23,884 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionRemoteClose closing a local session for connectionId[MF_83fe8b_1596565403336], entityName[mgmt-session], condition[null], description[null]
2020-08-04 14:06:23,884 DEBUG [com.azure.core.amqp.implementation.ReactorSession] Connection state: CLOSED
2020-08-04 14:06:23,884 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_83fe8b_1596565403336] sessionName[mgmt-session]: Complete. Removing and disposing session.
2020-08-04 14:06:23,885 INFO  [com.azure.core.amqp.implementation.ReactorSession] sessionId[mgmt-session]: Disposing of session.
2020-08-04 14:06:23,885 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionRemoteClose hostname[############.servicebus.windows.net:5671], connectionId[MF_83fe8b_1596565403336], errorCondition[amqp:connection:forced], errorDescription[The connection was inactive for more than the allowed 300000 milliseconds and is closed by container 'LinkTracker'. TrackingId:0d89a6f8acdc413996e0f60c7c7ce734_G20, SystemTracker:gateway5, Timestamp:2020-08-04T19:06:23]
2020-08-04 14:06:23,886 DEBUG [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_83fe8b_1596565403336]: Connection state: CLOSED
2020-08-04 14:06:23,886 INFO  [com.azure.messaging.eventhubs.implementation.EventHubConnectionProcessor] namespace[############.servicebus.windows.net] entityPath[testhub1]: Channel is closed.
2020-08-04 14:06:23,886 INFO  [com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection] connectionId[MF_83fe8b_1596565403336]: Disposing of connection.
2020-08-04 14:06:23,887 INFO  [com.azure.core.amqp.implementation.ReactorConnection] connectionId[MF_83fe8b_1596565403336]: Disposing of ReactorConnection.
2020-08-04 14:06:23,887 INFO  [com.azure.core.amqp.implementation.AmqpExceptionHandler] Shutdown received: ReactorExecutor.close() was called., isTransient[false], initiatedByClient[true]
2020-08-04 14:07:23,889 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] Unable to acquire dispose reactor semaphore within timeout.
2020-08-04 14:07:23,889 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onTransportClosed hostname[############.servicebus.windows.net:5671], connectionId[MF_83fe8b_1596565403336], error[n/a]
2020-08-04 14:07:23,890 INFO  [com.azure.core.amqp.implementation.handler.CustomIOHandler] onTransportClosed connectionId[MF_83fe8b_1596565403336], hostname[############.servicebus.windows.net:5671]
2020-08-04 14:07:23,890 DEBUG [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionLocalClose connectionId[mgmt-session], entityName[MF_83fe8b_1596565403336], condition[Error{condition=null, description='null', info=null}]
2020-08-04 14:07:23,891 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionLocalClose hostname[############.servicebus.windows.net:5671], connectionId[MF_83fe8b_1596565403336], errorCondition[null], errorDescription[null]
2020-08-04 14:07:23,891 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionUnbound hostname[############.servicebus.windows.net:5671], connectionId[MF_83fe8b_1596565403336], state[CLOSED], remoteState[CLOSED]
2020-08-04 14:07:23,891 INFO  [com.azure.core.amqp.implementation.handler.SendLinkHandler] onLinkFinal connectionId[MF_83fe8b_1596565403336],  linkName[mgmt:sender]
2020-08-04 14:07:23,892 INFO  [com.azure.core.amqp.implementation.handler.ReceiveLinkHandler] onLinkFinal connectionId[MF_83fe8b_1596565403336],  linkName[mgmt:receiver]
2020-08-04 14:07:23,892 INFO  [com.azure.core.amqp.implementation.handler.SessionHandler] onSessionFinal connectionId[MF_83fe8b_1596565403336], entityName[mgmt-session], condition[null], description[null]
2020-08-04 14:07:23,893 INFO  [com.azure.core.amqp.implementation.handler.ConnectionHandler] onConnectionFinal hostname[############.servicebus.windows.net:5671], connectionId[MF_83fe8b_1596565403336], errorCondition[null], errorDescription[null]
2020-08-04 14:07:23,913 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_83fe8b_1596565403336], message[Processing all pending tasks and closing old reactor.]
2020-08-04 14:07:23,915 INFO  [com.azure.core.amqp.implementation.ReactorExecutor] connectionId[MF_83fe8b_1596565403336], message[Stopping the reactor because thread was interrupted or the reactor has no more events to process.]

To Reproduce
Start an event hub processor client and do not publish any events to the hub.

Code Snippet
Just use the standard example

Expected behavior
Have the EventProcessorClient recover and start receiving events again after the 30 minute timeout. At the very least, provide a way to handle the exception.

Setup (please complete the following information):

  • OS: WIndows
  • IDE : Eclipse
  • Version of the Library used: Event Hubs 5.1.2 and checkpointstore-blob 1.1.2

Additional context
I think this might be directly related to #12397
@ghost ghost added needs-triage Workflow: This is a new issue that needs to be triaged to the appropriate team. customer-reported Issues that are reported by GitHub users external to the Azure organization. question The issue doesn't require a change to the product in order to be resolved. Most issues start as that labels Aug 4, 2020
@joshfree joshfree added bug This issue requires a change to an existing behavior in the product in order to be resolved. Client This issue points to a problem in the data-plane of the library. Event Hubs pillar-reliability The issue is related to reliability, one of our core engineering pillars. (includes stress testing) and removed question The issue doesn't require a change to the product in order to be resolved. Most issues start as that labels Aug 4, 2020
@ghost ghost removed the needs-triage Workflow: This is a new issue that needs to be triaged to the appropriate team. label Aug 4, 2020
@joshfree
Copy link
Member

joshfree commented Aug 4, 2020

Thanks for filing this Event Hubs issue @jonjarvis; @srnagar can you please investigate and follow up?

@srnagar
Copy link
Member

srnagar commented Aug 10, 2020

This issue is related to #13420

@conniey conniey mentioned this issue Sep 9, 2020
Merged
This was referenced Sep 14, 2020
Closed
Closed
@github-actions github-actions bot locked and limited conversation to collaborators Apr 12, 2023
Sign up for free to subscribe to this conversation on GitHub. Already have an account? Sign in.
Assignees

@srnagar srnagar

Labels
bug This issue requires a change to an existing behavior in the product in order to be resolved. Client This issue points to a problem in the data-plane of the library. customer-reported Issues that are reported by GitHub users external to the Azure organization. Event Hubs pillar-reliability The issue is related to reliability, one of our core engineering pillars. (includes stress testing)
Projects
None yet
Milestone
No milestone
Development

Successfully merging a pull request may close this issue.

Force closes links when connection is closed so it requests another link. conniey/azure-sdk-for-java
3 participants
@joshfree @jonjarvis @srnagar