Spring-cloud-sleuth: Error sending STOMP message via websocket

[purple52]

While upgrading from Spring Boot 2.0.5 to 2.1.1, I encountered a problem in a service which sends STOMP messages via a web socket. I have isolated the problem and created an example project:

https://github.com/purple52/websocket-sleuth-test

If you build that project and run the test using ./gradlew clean build --info, it'll fail with this error:

2019-01-17 10:01:14.876 ERROR [websocket-sleuth-test,86b7bbc97c2ae296,14d3361a3e31673e,false] 6022 --- [    Test worker] o.s.m.s.b.SimpleBrokerMessageHandler     : Failed to send GenericMessage [payload=byte[16], headers={spanTraceId=86b7bbc97c2ae296, spanId=86b7bbc97c2ae296, simpMessageType=MESSAGE, nativeHeaders={spanTraceId=[86b7bbc97c2ae296], spanId=[86b7bbc97c2ae296], spanSampled=[0]}, spanSampled=0, contentType=application/json;charset=UTF-8, simpDestination=/topic/foo}]
2019-01-17 10:01:14.877 ERROR [websocket-sleuth-test,86b7bbc97c2ae296,86b7bbc97c2ae296,false] 6022 --- [    Test worker] o.s.m.s.ExecutorSubscribableChannel      : Exception from afterMessageHandled in org.springframework.cloud.sleuth.instrument.messaging.TracingChannelInterceptor@3d415f8

I am not 100% sure this is a Cloud Sleuth defect, or whether it's Spring Messaging at fault, but I'm starting by reporting it here, since my test passes if I remove Cloud Sleuth from the project. This is using version 2.0.2-RELEASE of Cloud Sleuth, but still happens if I override and use 2.1.0-RC3.

What appears to happen is that TracingChannelInterceptor tries to remove the trace headers by calling NativeMessageHeaderAccessor.removeNativeHeader which checks that the headers are mutable, then tries to remove the native header. However, although the accessor says it is mutable, the native headers are returned in an unmodifiable map which throws an UnsupportedOperationException when the attempt is made to remove the headers.

You can see the full exception thrown with a breakpoint at AbstractMessageChannel.java#L142. Here's the stacktrace:

java.lang.UnsupportedOperationException
    at java.util.Collections$UnmodifiableMap.remove(Collections.java:1460)
    at org.springframework.messaging.support.NativeMessageHeaderAccessor.removeNativeHeader(NativeMessageHeaderAccessor.java:209)
    at org.springframework.cloud.sleuth.instrument.messaging.MessageHeaderPropagation.removeAnyTraceHeaders(MessageHeaderPropagation.java:158)
    at org.springframework.cloud.sleuth.instrument.messaging.TracingChannelInterceptor.preSend(TracingChannelInterceptor.java:135)
    at org.springframework.messaging.support.AbstractMessageChannel$ChannelInterceptorChain.applyPreSend(AbstractMessageChannel.java:178)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:132)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:122)
    at org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.lambda$sendMessageToSubscribers$0(SimpleBrokerMessageHandler.java:401)
    at java.util.Map.forEach(Map.java:630)
    at org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.sendMessageToSubscribers(SimpleBrokerMessageHandler.java:388)
    at org.springframework.messaging.simp.broker.SimpleBrokerMessageHandler.handleMessageInternal(SimpleBrokerMessageHandler.java:304)
    at org.springframework.messaging.simp.broker.AbstractBrokerMessageHandler.handleMessage(AbstractBrokerMessageHandler.java:256)
    at org.springframework.messaging.support.ExecutorSubscribableChannel$SendTask.run(ExecutorSubscribableChannel.java:144)
    at org.springframework.messaging.support.ExecutorSubscribableChannel.sendInternal(ExecutorSubscribableChannel.java:100)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:136)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:122)
    at org.springframework.messaging.simp.SimpMessagingTemplate.sendInternal(SimpMessagingTemplate.java:187)
    at org.springframework.messaging.simp.SimpMessagingTemplate.doSend(SimpMessagingTemplate.java:162)
    at org.springframework.messaging.simp.SimpMessagingTemplate.doSend(SimpMessagingTemplate.java:48)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:109)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:151)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:129)
    at org.springframework.messaging.core.AbstractMessageSendingTemplate.convertAndSend(AbstractMessageSendingTemplate.java:122)
    at com.example.WebsocketTest.Can pass message via websocket(WebsocketTest.kt:69)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
    at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
    at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
    at org.springframework.test.context.junit4.statements.RunBeforeTestExecutionCallbacks.evaluate(RunBeforeTestExecutionCallbacks.java:74)
    at org.springframework.test.context.junit4.statements.RunAfterTestExecutionCallbacks.evaluate(RunAfterTestExecutionCallbacks.java:84)
    at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75)
    at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86)
    at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84)
    at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:251)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
    at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
    at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
    at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
    at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
    at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
    at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
    at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
    at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
    at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
    at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)

[marcingrzejszczak]

I had a chat about this with @artembilan and he pointed that maybe @rstoyanchev can have more to say about this.

It's strange (at least for me) that native headers are mutable, but then turn out to be immutable.

[rstoyanchev]

TracingChannelInterceptor uses getMutableAccessor and it might already be that, or otherwise a new instance is created. Maybe it's the latter and it doesn't do a deep copy of the (immutable) native headers? So that could be a bug (spring-messaging).

But why did become immutable in the first place? The simple broker sends it mutable and it's only the last interceptor on the channel that should flip it immutable. Perhaps some other interceptor is not using getMutableAccessor like TracingChannelInterceptor does?

[marcingrzejszczak]

So if I understand it correctly, the problem is not in how Sleuth handles the accessor, but some other interceptor must be not using it properly? Do you know who is working on spring-messaging and could help me debug this?

[rstoyanchev]

I can help but not today. It could be two problems. 1) some interceptor makes the headers immutable. you'll see in the stacktrace where AbstractMessageChannel calls the interceptor. Debug to see what other interceptors and maybe check after each to see when it becomes immutable. 2) even if it becomes immutable and then mutable again, it should work so there might be a bug in spring-messaging too.

[marcingrzejszczak]

OK thanks for the hints. Now I understand the problem, and if I have some spare time I'll debug this for sure.

[larsduelfer]

Any news on this? I am currently facing the same issue.

[tusciucalecs]

+1

[carlosvaztec]

I am wondering if it could be related to this line: https://github.com/spring-projects/spring-framework/blob/master/spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessagingTemplate.java#L161

[sunnykaka]

I encountered the same problem.
Is there any workaround to fix the problem?

[marcingrzejszczak]

I'm working on this as we speak. The workaround is to disable Sleuth's websocket messaging support via spring.sleuth.integration.websockets.enabled=false

[rstoyanchev]

I believe the root cause for this to be https://github.com/spring-projects/spring-framework/issues/25821 which means the workaround made here should be possible to remove to make the code more efficient. I'll open a separate issue for that.

[rstoyanchev]

On further thought the workaround is conditional so removing it won't make it more efficient, but at least a code level comment could be added to the workaround linking to the Spring Framework issue, and eventually the workaround could be removed.

[sanjaygverma]

I see still this error in version with spring boot 2.3.2 as well.

[rstoyanchev]

@sanjaygverma, if you take a look at the linked issue, the fix is in Spring Framework 5.2.10 while Spring Boot 2.3.2 depends on 5.2.8. So you don't have the fix.

[sanyarnd]

The issue is still present on 2.3.5 and 2.4.0-RC1 with Hoxton.SR8 and 2020-M3 cloud dependencies.

Stacktrace:

o.s.w.s.m.StompSubProtocolHandler        : Failed to send client message to application via MessageChannel in session ee3f0b36-30cf-232e-802c-d645e7096ccb. Sending STOMP ERROR to client.

org.springframework.messaging.MessageDeliveryException: Failed to send message to ExecutorSubscribableChannel[clientInboundChannel]; nested exception is java.util.ConcurrentModificationException
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:149)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:125)
    at org.springframework.web.socket.messaging.StompSubProtocolHandler.handleMessageFromClient(StompSubProtocolHandler.java:306)
    at org.springframework.web.socket.messaging.SubProtocolWebSocketHandler.handleMessage(SubProtocolWebSocketHandler.java:335)
    at org.springframework.web.socket.handler.WebSocketHandlerDecorator.handleMessage(WebSocketHandlerDecorator.java:75)
    at org.springframework.web.socket.handler.LoggingWebSocketHandlerDecorator.handleMessage(LoggingWebSocketHandlerDecorator.java:56)
    at org.springframework.web.socket.handler.ExceptionWebSocketHandlerDecorator.handleMessage(ExceptionWebSocketHandlerDecorator.java:58)
    at org.springframework.web.socket.adapter.standard.StandardWebSocketHandlerAdapter.handleTextMessage(StandardWebSocketHandlerAdapter.java:114)
    at org.springframework.web.socket.adapter.standard.StandardWebSocketHandlerAdapter.access$000(StandardWebSocketHandlerAdapter.java:43)
    at org.springframework.web.socket.adapter.standard.StandardWebSocketHandlerAdapter$3.onMessage(StandardWebSocketHandlerAdapter.java:85)
    at org.springframework.web.socket.adapter.standard.StandardWebSocketHandlerAdapter$3.onMessage(StandardWebSocketHandlerAdapter.java:82)
    at org.apache.tomcat.websocket.WsFrameBase.sendMessageText(WsFrameBase.java:402)
    at org.apache.tomcat.websocket.server.WsFrameServer.sendMessageText(WsFrameServer.java:119)
    at org.apache.tomcat.websocket.WsFrameBase.processDataText(WsFrameBase.java:502)
    at org.apache.tomcat.websocket.WsFrameBase.processData(WsFrameBase.java:301)
    at org.apache.tomcat.websocket.WsFrameBase.processInputBuffer(WsFrameBase.java:133)
    at org.apache.tomcat.websocket.server.WsFrameServer.onDataAvailable(WsFrameServer.java:82)
    at org.apache.tomcat.websocket.server.WsFrameServer.doOnDataAvailable(WsFrameServer.java:171)
    at org.apache.tomcat.websocket.server.WsFrameServer.notifyDataAvailable(WsFrameServer.java:151)
    at org.apache.tomcat.websocket.server.WsHttpUpgradeHandler.upgradeDispatch(WsHttpUpgradeHandler.java:148)
    at org.apache.coyote.http11.upgrade.UpgradeProcessorInternal.dispatch(UpgradeProcessorInternal.java:54)
    at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:59)
    at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:868)
    at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1590)
    at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630)
    at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61)
    at java.base/java.lang.Thread.run(Thread.java:832)
Caused by: java.util.ConcurrentModificationException: null
    at java.base/java.util.HashMap$HashIterator.nextNode(HashMap.java:1584)
    at java.base/java.util.HashMap$EntryIterator.next(HashMap.java:1617)
    at java.base/java.util.HashMap$EntryIterator.next(HashMap.java:1615)
    at java.base/java.util.Collections$UnmodifiableMap$UnmodifiableEntrySet$1.next(Collections.java:1679)
    at java.base/java.util.Collections$UnmodifiableMap$UnmodifiableEntrySet$1.next(Collections.java:1672)
    at java.base/java.util.Map.forEach(Map.java:651)
    at org.springframework.messaging.support.MessageHeaderAccessor.copyHeaders(MessageHeaderAccessor.java:382)
    at org.springframework.messaging.support.NativeMessageHeaderAccessor.copyHeaders(NativeMessageHeaderAccessor.java:135)
    at org.springframework.cloud.sleuth.instrument.messaging.TracingChannelInterceptor.outputMessage(TracingChannelInterceptor.java:224)
    at org.springframework.cloud.sleuth.instrument.messaging.TracingChannelInterceptor.preSend(TracingChannelInterceptor.java:191)
    at org.springframework.messaging.support.AbstractMessageChannel$ChannelInterceptorChain.applyPreSend(AbstractMessageChannel.java:181)
    at org.springframework.messaging.support.AbstractMessageChannel.send(AbstractMessageChannel.java:135)
    ... 28 common frames omitted

[sanyarnd]

Here's a demo project + HTML sample

<html lang="ru">
<head>
    <title>WebSocket demo</title>
    <script src="https://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.min.js"
            integrity="sha512-iKDtgDyTHjAitUDdLljGhenhPwrbBfqTKWO1mkhSFH3A7blITC9MhYon6SjnMhp4o0rADGw9yAC6EW4t5a4K3g=="
            crossorigin="anonymous"></script>
    <script type="text/javascript">
        let stompClient = undefined;

        function connect() {
            const url = "ws://localhost:8080/ws";
            stompClient = Stomp.client(url);

            stompClient.connect({   }, function (frame) {
                stompClient.subscribe("/messages/0", function (message) {
                });
            });
        }
    </script>
</head>
<div>
    <div>
        <button id="connect" onclick="connect();">Connect</button>
    </div>
</div>
</html>

websocket-sleuth-cloud.zip

[tuomoa]

We are getting the same error after upgrading to Spring Boot 2.3.5 (with Hoxton.SR7 or Hoxton.SR8). Spring Boot 2.3.4 is working with Hoxton.SR7.

[ivangfr]

I've faced the same problem when I updated my project springboot-kafka-websocket

• from spring-boot version 2.3.3.RELEASE to 2.4.0
• from spring-cloud version Hoxton.SR7 to 2020.0.0-M6.

The fix, for now, is what @marcingrzejszczak suggested, spring.sleuth.integration.websockets.enabled=false

[marcingrzejszczak]

Hey @ivangfr I've managed to run your sample (more or less) with Cloud release train 2020.0.0-SNAPSHOT and Sleuth with websockets turned on (Boot is of course in 2.4.0). I got the following trace in Zipkin and no exceptions in the logs. Can you double check yourself that the latest snapshots work fine?

[ivangfr]

Hi @marcingrzejszczak I've just tried the Cloud release 2020.0.0-SNAPSHOT and the issue is still there.

Btw, sorry for not detailing how to check the issue in my project.

If you have the springboot-kafka-websocket apps running, access bitcoin-client app at http://localhost:9082/ and provide the credentials: username: user1, password: 123.

You will see, using the browser inspect, that the websocket is not able to connect.

Then, set spring.sleuth.integration.websockets.enabled to false. This time, once the app is restarted, it will connect successfully.

[marcingrzejszczak]

OK, now I understand. Will check it again later.

[marcingrzejszczak]

I believe that now the issue is fixed. Can you please check it out against the latest snapshots (2.2.x or 3.0.x) ?

[ivangfr]

Hi @marcingrzejszczak ! Yes, it's fixed. Thanks! I will wait for the next Cloud 2020.0.0 release to remove the property spring.sleuth.integration.websockets.enabled=false.

[rstoyanchev]

This might be the same issue as https://github.com/spring-projects/spring-framework/issues/26155.