问题描述

使用Java SDK连接Azure Event Hub，一直出现 java.util.concurrent.TimeoutException 异常， 消息为：java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 60000ms in 'flatMapMany' (and no fallback has been configured)。

且消息体中并没有更多有效消息。

ERROR .e.r.OpportunityResourceEventhubReceiver []: com.cbs.message.facade.eventhub.receive.OpportunityResourceEventhubReceiver.onError.partition:NONE. Exception:{}
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:294)
    at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.doTimeout(FluxTimeout.java:279)
    at reactor.core.publisher.FluxTimeout$TimeoutTimeoutSubscriber.onNext(FluxTimeout.java:418)
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
    at reactor.core.publisher.MonoDelay$MonoDelayRunnable.propagateDelay(MonoDelay.java:270)
    at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:285)
    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)

问题解答

如果使用Azure Event Hub官方的SDK， 可以通过设置日志输出级别为Info来查看更多详细的日志，这样就可以查看更详细的日志输出。比如：

2022-11-03 10:57:49.410  INFO  ---  [           main] c.a.m.eventhubs.EventHubClientBuilder    []: {"az.sdk.message":"Emitting a single connection.","connectionId":"MF_22f21s_6940767444216"}
2022-11-03 10:57:49.602  INFO  ---  [           main] c.a.m.e.i.EventHubConnectionProcessor    []: {"az.sdk.message":"Setting next AMQP channel.","entityPath":"eh01"}
2022-11-03 10:57:49.603  INFO  ---  [           main] c.a.m.e.i.EventHubConnectionProcessor    []: {"az.sdk.message":"Next AMQP channel received, updating 0 current subscribers","entityPath":"eh01"}
2022-11-03 10:57:49.612  INFO  ---  [           main] c.a.m.eventhubs.EventProcessorClient     []: {"az.sdk.message":"Starting a new event processor instance.","eventProcessorId":"02690c22-21be-4b39-b976-efcf3ce3819a"}
2022-11-03 10:57:49.629  INFO  ---  [           main] c.a.m.eventhubs.EventHubClientBuilder    []: {"az.sdk.message":"Emitting a single connection.","connectionId":"MF_93d32o_12dcdedeadcfe33"}
2022-11-03 10:57:49.630  INFO  ---  [           main] c.a.m.e.i.EventHubConnectionProcessor    []: {"az.sdk.message":"Setting next AMQP channel.","entityPath":"eh02"}
2022-11-03 10:57:49.698  INFO  ---  [           main] c.a.m.e.i.EventHubConnectionProcessor    []: {"az.sdk.message":"Next AMQP channel received, updating 0 current subscribers","entityPath":"eh02"}
2022-11-03 10:57:49.699  INFO  ---  [           main] c.a.m.eventhubs.EventProcessorClient     []: {"az.sdk.message":"Starting a new event processor instance.","eventProcessorId":"c8d655c2-d12d-4d14-a85e-e333273293d9"}
2022-11-03 10:57:49.712  INFO  ---  [           main] c.a.m.eventhubs.EventHubClientBuilder    []: {"az.sdk.message":"Emitting a single connection.","connectionId":"MF_0346b3_1667444269712"}
2022-11-03 10:57:49.713  INFO  ---  [           main] c.a.m.e.i.EventHubConnectionProcessor    []: {"az.sdk.message":"Setting next AMQP channel.","entityPath":"eh02"}
2022-11-03 10:57:49.713  INFO  ---  [           main] c.a.m.e.i.EventHubConnectionProcessor    []: {"az.sdk.message":"Next AMQP channel received, updating 0 current subscribers","entityPath":"eh02"}
2022-11-03 10:57:49.714  INFO  ---  [           main] c.a.m.eventhubs.EventProcessorClient     []: {"az.sdk.message":"Starting a new event processor instance.","eventProcessorId":"2fd3a905-c39c-47ff-bc8d-e4b21301eeb3"}
2022-11-03 10:57:50.127  INFO  ---  [pool-6-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Starting load balancer.","ownerId":"02690c22-21be-4b39-b976-efcf3ce3819a"}
2022-11-03 10:57:50.136  INFO  ---  [pool-6-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Getting partitions from Event Hubs service.","entityPath":"eh01"}
2022-11-03 10:57:51.016  INFO  ---  [pool-6-thread-1] c.a.c.a.i.ReactorConnection              []: {"az.sdk.message":"Creating and starting connection.","connectionId":"MF_22f21s_6940767444216","hostName":"test-eventhub.servicebus.chinacloudapi.cn","port":5671}
2022-11-03 10:57:51.052  INFO  ---  [pool-6-thread-1] c.a.c.a.implementation.ReactorExecutor   []: {"az.sdk.message":"Starting reactor.","connectionId":"MF_22f21s_6940767444216"}
2022-11-03 10:57:51.060  INFO  ---  [ctor-executor-1] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionInit","connectionId":"MF_22f21s_6940767444216","hostName":"test-eventhub.servicebus.chinacloudapi.cn","namespace":"test-eventhub.servicebus.chinacloudapi.cn"}
2022-11-03 10:57:51.061  INFO  ---  [ctor-executor-1] c.a.c.a.i.handler.ReactorHandler         []: {"az.sdk.message":"reactor.onReactorInit","connectionId":"MF_22f21s_6940767444216"}
2022-11-03 10:57:51.061  INFO  ---  [ctor-executor-1] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionLocalOpen","connectionId":"MF_22f21s_6940767444216","errorCondition":null,"errorDescription":null,"hostName":"test-eventhub.servicebus.chinacloudapi.cn"}
2022-11-03 10:57:51.220  INFO  ---  [ctor-executor-1] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionBound","connectionId":"MF_22f21s_6940767444216","hostName":"test-eventhub.servicebus.chinacloudapi.cn","peerDetails":"test-eventhub.servicebus.chinacloudapi.cn:5671"}
2022-11-03 10:57:51.257  INFO  ---  [ctor-executor-1] c.a.c.a.i.handler.StrictTlsContextSpi    []: SSLv2Hello was an enabled protocol. Filtering out.
2022-11-03 10:57:51.360  INFO  ---  [pool-8-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Starting load balancer.","ownerId":"2fd3a905-c39c-47ff-bc8d-e4b21301eeb3"}
2022-11-03 10:57:51.360  INFO  ---  [pool-8-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Getting partitions from Event Hubs service.","entityPath":"eh02"}
2022-11-03 10:57:51.361  INFO  ---  [pool-8-thread-1] c.a.c.a.i.ReactorConnection              []: {"az.sdk.message":"Creating and starting connection.","connectionId":"MF_0346b3_1667444269712","hostName":"test-eventhub.servicebus.chinacloudapi.cn","port":5671}
2022-11-03 10:57:51.363  INFO  ---  [pool-8-thread-1] c.a.c.a.implementation.ReactorExecutor   []: {"az.sdk.message":"Starting reactor.","connectionId":"MF_0346b3_1667444269712"}
2022-11-03 10:57:51.364  INFO  ---  [ctor-executor-2] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionInit","connectionId":"MF_0346b3_1667444269712","hostName":"test-eventhub.servicebus.chinacloudapi.cn","namespace":"test-eventhub.servicebus.chinacloudapi.cn"}
2022-11-03 10:57:51.366  INFO  ---  [ctor-executor-2] c.a.c.a.i.handler.ReactorHandler         []: {"az.sdk.message":"reactor.onReactorInit","connectionId":"MF_0346b3_1667444269712"}
2022-11-03 10:57:51.366  INFO  ---  [ctor-executor-2] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionLocalOpen","connectionId":"MF_0346b3_1667444269712","errorCondition":null,"errorDescription":null,"hostName":"test-eventhub.servicebus.chinacloudapi.cn"}
2022-11-03 10:57:51.368  INFO  ---  [ctor-executor-2] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionBound","connectionId":"MF_0346b3_1667444269712","hostName":"test-eventhub.servicebus.chinacloudapi.cn","peerDetails":"test-eventhub.servicebus.chinacloudapi.cn:5671"}
2022-11-03 10:57:51.369  INFO  ---  [ctor-executor-2] c.a.c.a.i.handler.StrictTlsContextSpi    []: SSLv2Hello was an enabled protocol. Filtering out.
2022-11-03 10:57:51.470  INFO  ---  [pool-7-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Starting load balancer.","ownerId":"c8d655c2-d12d-4d14-a85e-e333273293d9"}
2022-11-03 10:57:51.470  INFO  ---  [pool-7-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Getting partitions from Event Hubs service.","entityPath":"eh02"}
2022-11-03 10:57:51.471  INFO  ---  [pool-7-thread-1] c.a.c.a.i.ReactorConnection              []: {"az.sdk.message":"Creating and starting connection.","connectionId":"MF_93d32o_12dcdedeadcfe33","hostName":"test-eventhub.servicebus.chinacloudapi.cn","port":5671}
2022-11-03 10:57:51.472  INFO  ---  [pool-7-thread-1] c.a.c.a.implementation.ReactorExecutor   []: {"az.sdk.message":"Starting reactor.","connectionId":"MF_93d32o_12dcdedeadcfe33"}
2022-11-03 10:57:51.473  INFO  ---  [ctor-executor-3] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionInit","connectionId":"MF_93d32o_12dcdedeadcfe33","hostName":"test-eventhub.servicebus.chinacloudapi.cn","namespace":"test-eventhub.servicebus.chinacloudapi.cn"}
2022-11-03 10:57:51.473  INFO  ---  [ctor-executor-3] c.a.c.a.i.handler.ReactorHandler         []: {"az.sdk.message":"reactor.onReactorInit","connectionId":"MF_93d32o_12dcdedeadcfe33"}
2022-11-03 10:57:51.473  INFO  ---  [ctor-executor-3] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionLocalOpen","connectionId":"MF_93d32o_12dcdedeadcfe33","errorCondition":null,"errorDescription":null,"hostName":"test-eventhub.servicebus.chinacloudapi.cn"}
2022-11-03 10:57:51.475  INFO  ---  [ctor-executor-3] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionBound","connectionId":"MF_93d32o_12dcdedeadcfe33","hostName":"test-eventhub.servicebus.chinacloudapi.cn","peerDetails":"test-eventhub.servicebus.chinacloudapi.cn:5671"}
2022-11-03 10:57:51.475  INFO  ---  [ctor-executor-3] c.a.c.a.i.handler.StrictTlsContextSpi    []: SSLv2Hello was an enabled protocol. Filtering out.
2022-11-03 10:58:21.099  INFO  ---  [pool-6-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: Load balancer already running
2022-11-03 10:58:21.367  INFO  ---  [pool-8-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: Load balancer already running
2022-11-03 10:58:21.474  INFO  ---  [pool-7-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: Load balancer already running
2022-11-03 10:58:51.014  WARN  ---  [     parallel-2] c.a.m.e.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)
2022-11-03 10:58:51.017 ERROR  ---  [     parallel-2] c.c.m.f.e.r.OpportunityEventhubReceiver  []: com.cbs.message.facade.eventhub.receive.OpportunityEventhubReceiver.onError.partition:NONE. Exception:{}
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:294)
    at reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.doTimeout(FluxTimeout.java:279)
    at reactor.core.publisher.FluxTimeout$TimeoutTimeoutSubscriber.onNext(FluxTimeout.java:418)
    at reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:79)
    at reactor.core.publisher.MonoDelay$MonoDelayRunnable.propagateDelay(MonoDelay.java:270)
    at reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:285)
    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)
2022-11-03 10:58:51.058  WARN  ---  [ctor-executor-1] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onTransportError","connectionId":"MF_93d32o_12dcdedeadcfe33","errorCondition":"amqp:connection:framing-error","errorDescription":"org.apache.qpid.proton.engine.TransportException: connection aborted","hostName":"test-eventhub.servicebus.chinacloudapi.cn"}
2022-11-03 10:58:51.062  INFO  ---  [ctor-executor-1] c.a.c.a.i.ReactorConnection              []: {"az.sdk.message":"Disposing of ReactorConnection.","connectionId":"MF_93d32o_12dcdedeadcfe33","isTransient":false,"isInitiatedByClient":false,"shutdownMessage":"org.apache.qpid.proton.engine.TransportException: connection aborted, errorContext[NAMESPACE: test-eventhub.servicebus.chinacloudapi.cn. ERROR CONTEXT: N/A]"}
2022-11-03 10:58:51.075  INFO  ---  [ctor-executor-1] c.a.c.a.i.handler.ConnectionHandler      []: {"az.sdk.message":"onConnectionUnbound","connectionId":"MF_93d32o_12dcdedeadcfe33","hostName":"test-eventhub.servicebus.chinacloudapi.cn","state":"CLOSED","remoteState":"UNINITIALIZED"}
2022-11-03 10:58:51.077  INFO  ---  [ctor-executor-1] c.a.c.a.i.ReactorConnection              []: {"az.sdk.message":"Closing executor.","connectionId":"MF_93d32o_12dcdedeadcfe33"}
2022-11-03 10:58:51.099  INFO  ---  [pool-6-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Starting load balancer.","ownerId":"02690c22-21be-4b39-b976-efcf3ce3819a"}
2022-11-03 10:58:51.100  INFO  ---  [pool-6-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: {"az.sdk.message":"Getting partitions from Event Hubs service.","entityPath":"eh02"}
2022-11-03 10:58:51.101 ERROR  ---  [pool-6-thread-1] c.a.m.e.i.EventHubReactorAmqpConnection  []: {"az.sdk.message":"connectionId[MF_93d32o_12dcdedeadcfe33]: Connection is disposed. Cannot get management instance","exception":"connectionId[MF_22f21s_6940767444216]: Connection is disposed. Cannot get management instance","connectionId":"MF_22f21s_6940767444216"}
2022-11-03 10:58:51.104  WARN  ---  [pool-6-thread-1] c.a.m.e.PartitionBasedLoadBalancer       []: Load balancing for event processor failed.connectionId[MF_93d32o_12dcdedeadcfe33]: Connection is disposed. Cannot get management instance
2022-11-03 10:58:51.105 ERROR  ---  [pool-6-thread-1] c.c.m.f.e.r.OpportunityEventhubReceiver  []: com.cbs.message.facade.eventhub.receive.OpportunityEventhubReceiver.onError.partition:NONE. Exception:{}
java.lang.IllegalStateException: connectionId[MF_22f21s_6940767444216]: Connection is disposed. Cannot get management instance
    at com.azure.messaging.eventhubs.implementation.EventHubReactorAmqpConnection.getManagementNode(EventHubReactorAmqpConnection.java:90)
    at com.azure.messaging.eventhubs.EventHubAsyncClient.lambda$getProperties$0(EventHubAsyncClient.java:73)
    at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:125)
    at reactor.core.publisher.Operators$MonoSubscriber.complete(Operators.java:1815)
    at com.azure.core.amqp.implementation.AmqpChannelProcessor.subscribe(AmqpChannelProcessor.java:267)
    at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
    at reactor.core.publisher.MonoZip.subscribe(MonoZip.java:128)
    at reactor.core.publisher.Mono.subscribe(Mono.java:4150)
    at reactor.core.publisher.MonoIgnorePublisher.subscribe(MonoIgnorePublisher.java:56)
    at reactor.core.publisher.FluxRepeatPredicate$RepeatPredicateSubscriber.resubscribe(FluxRepeatPredicate.java:119)
    at reactor.core.publisher.MonoRepeatPredicate.subscribeOrReturn(MonoRepeatPredicate.java:47)
    at reactor.core.publisher.Flux.subscribe(Flux.java:8221)
    at reactor.core.publisher.Flux.subscribeWith(Flux.java:8408)
    at reactor.core.publisher.Flux.subscribe(Flux.java:8205)
    at reactor.core.publisher.Flux.subscribe(Flux.java:8129)
    at com.azure.messaging.eventhubs.PartitionBasedLoadBalancer.loadBalance(PartitionBasedLoadBalancer.java:154)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    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)

根据以上的日志，可以看出：

1）在发生异常之前，有WARN日志输出，显示 Load balancing for event processor failed.

2）在连接的信息中，查看到应用连接的端口为 5671

由于一直连接不上，就应该参考官网的客户端连接问题排查访问进行排查，第一步就是检查端口 5671 是否能从执行 Java代码的主机上Ping 通。参考：https://docs.azure.cn/zh-cn/event-hubs/troubleshooting-guide#run-the-command-to-check-dropped-packets

请运行以下命令，检查是否存在任何丢弃的数据包或者无法ping通的问题：

.\psping.exe -n 25 -i 1 -q <yournamespacename>.servicebus.chinacloudapi.cn:5671 -nobanner    

在实际验证中，以上错误就是发现 客户端环境无法ping通 5671，5672端口。当在防火墙中放开这两个端口后。 连接成功！

参考资料

排查连接问题 - Azure 事件中心： https://docs.azure.cn/zh-cn/event-hubs/troubleshooting-guide#run-the-command-to-check-dropped-packets

Load balancing for event processor failed ****#12525 : https://github.com/Azure/azure-sdk-for-java/issues/12525

image.png

当在复杂的环境中面临问题，格物之道需：浊而静之徐清，安以动之徐生。 云中，恰是如此!

分类: 【Azure 事件中心】

标签: Did not observe any item or terminal signal, within 60000ms in 'flatMapMany', java.util.concurrent.TimeoutException