Azure / Azure/azure-service-bus-java
Internal receive link of requestresponselink to '$cbs' closed with error.
- Dominant language
- Java
- Stars
- 62
- Forks
- 63
- PR merge metrics
- No merged PRs in 30d
Description
As part of a high volume application we are running 4 workers consuming the IOT Hub and passing messages along to the service bus. These workers are deployed in an AKS cluster, where each worker consumes one partition in the IOT Hub.
To preserve connecting overhead we cache the IMessageSender in a local concurrent map per queue and re-use them for every message. However after some time some of the workers are throwing exceptions about a $sbc link being closed after being inactive for 60000ms. Which is strange since there is a constant flow of messages to the service bus.
What we end up with is similar to this:
```java
IMessageSender cachedSender = ClientFactory.createMessageSenderFromConnectionStringBuilder(connectionStringBuilder);
// on every message we receive from the IOTHub we split up the message and send multiple messages to the service bus like this
cachedSender.sendAsync(convertMessage(message, partitionKey))
.exceptionally(th -> {
this.errorCount++;
throw new AmqpException("Could not deliver message to destination, exception received from Azure", th);
})
.thenRun(() -> errorCount = 0)
```
## Actual Behavior
1. After some time sending messages using sendAsync exceptions in the transport layer occur
## Expected Behavior
1. All messages are sent to the service bus through the sendAsync
## Versions
- Linux OS (Java 8 openjdk)
- Version: 2.0.0-preview-3 (also tried 1.2.6)
Stacktrace produced by the java service bus SDK:
```
18-07-10 13:16:22.577 WARN c.m.a.s.primitives.RequestResponseLink : Completing all pending requests with exception in request response link to $cbs
2018-07-10 13:16:22.579 WARN c.m.a.s.primitives.RequestResponseLink : Internal receive link of requestresponselink to '$cbs' closed with error.
com.microsoft.azure.servicebus.primitives.ServiceBusException: Error{condition=amqp:connection:forced, description='The connection was inactive for more than the allowed 60000 milliseconds and is closed by container 'LinkTracker'. TrackingId:ad8af2f1f61b400c9ed9b0e1c1f931fe_G19, SystemTracker:gateway7, Timestamp:7/10/2018 1:16:22 PM', info=null}
at com.microsoft.azure.servicebus.primitives.ExceptionUtil.toException(ExceptionUtil.java:113)
at com.microsoft.azure.servicebus.primitives.RequestResponseLink$InternalReceiver.onClose(RequestResponseLink.java:661)
at com.microsoft.azure.servicebus.amqp.BaseLinkHandler.processOnClose(BaseLinkHandler.java:68)
at com.microsoft.azure.servicebus.amqp.BaseLinkHandler.onLinkRemoteClose(BaseLinkHandler.java:42)
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.microsoft.azure.servicebus.primitives.MessagingFactory$RunReactor.run(MessagingFactory.java:559)
at java.lang.Thread.run(Thread.java:748)
```
Contributor guide
Assessment
This issue has not been assessed yet.