wso2esb + websocket - ConcurrentModificationException error

Question

I am using wso2esb + wso2mb + websockets to transform JMS messages from wso2mb to websockets. During my performance tests (I have tried to send 7k messages from wso2mb to websocket) I got following errror message:

java.util.ConcurrentModificationException
    at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:901)
    at java.util.ArrayList$Itr.next(ArrayList.java:851)
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.management.WebsocketSubscriberPathManager.broadcastOnSubscriberPath(WebsocketSubscriberPathManager.java:98)
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketResponseSender.handleSendBack(InboundWebsocketResponseSender.java:117)
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketResponseSender.sendBack(InboundWebsocketResponseSender.java:86)
    at org.apache.synapse.core.axis2.Axis2Sender.sendBack(Axis2Sender.java:214)
    at org.apache.synapse.mediators.builtin.RespondMediator.mediate(RespondMediator.java:35)
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:97)
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:59)
    at org.apache.synapse.config.xml.AnonymousListMediator.mediate(AnonymousListMediator.java:37)
    at org.apache.synapse.config.xml.SwitchCase.mediate(SwitchCase.java:69)
    at org.apache.synapse.mediators.filters.SwitchMediator.mediate(SwitchMediator.java:119)
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:97)
    at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:59)
    at org.apache.synapse.mediators.base.SequenceMediator.mediate(SequenceMediator.java:158)
    at org.apache.synapse.core.axis2.Axis2SynapseEnvironment.injectMessage(Axis2SynapseEnvironment.java:993)
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.injectToSequence(InboundWebsocketSourceHandler.java:461)
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.handleWebsocketPassthroughTextFrame(InboundWebsocketSourceHandler.java:346)
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.handleWebSocketFrame(InboundWebsocketSourceHandler.java:242)
    at org.wso2.carbon.inbound.endpoint.protocol.websocket.InboundWebsocketSourceHandler.channelRead(InboundWebsocketSourceHandler.java:132)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at io.netty.handler.codec.MessageToMessageDecoder.channelRead(MessageToMessageDecoder.java:103)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:244)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:110)
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
    at java.lang.Thread.run(Thread.java:745)

Do you know what could be the reason? Let me know if you would like to see my wso2esb project.

Thank you very much!

wso2 mb does not support web sockets at the moment. Can you please explain more what you tried to achieve and attach whole project if possible ? — plr, Oct 03 '16 at 03:13

score 1 · Answer 1 · answered Oct 03 '16 at 10:35

When new clients are connected, it is added to a map. If you broadcast during a load test, the same map is used to add connections and for broadcasting purposes. This will ended up with a Concurrent Modification Exception. This is a known behavior and we recommend you to stop broadcasting while you are running performance tests. Instead write only to one channel during the performance tests and that would solve this issue. You man find the code at [1] for further analysis.

[1] https://github.com/wso2/carbon-mediation/blob/master/components/inbound-endpoints/org.wso2.carbon.inbound.endpoint/src/main/java/org/wso2/carbon/inbound/endpoint/protocol/websocket/management/WebsocketSubscriberPathManager.java#L98

I noticed that when I am using broadcasting (`ws.client.side.broadcast.level=1`), the map (called `inboundSubscriberPathMap`) in class `WebsocketSubscriberPathManager` is cleared up only when I restarted the WSO2 ESB. Is there any way to "close" or "clear" somehow subscriptions that are no more valid? Let assume that I have connected to the websocket from front-end application to wso2, then wso2 cached for me subscription on specific path, but when I will close the connection then wso2 is not removing subscription from `inboundSubscriberPathMap`. How to avoid it? — bontade, Oct 04 '16 at 12:39

wso2esb + websocket - ConcurrentModificationException error

1 Answers1