Frequent send to spring-websocket session: lost in transit

Question

I got a load-test setup of spring websocket server (based on Jetty and spring version 4.3.2.RELEASE) and client, that generates many connections (based on spring's sample java websocket client). The code below sends data to given websocket session: the snippet exploits the case where sessionId can be used instead of User ID (Spring WebSocket @SendToSession: send message to specific session). I may execute this code very often, every 2-3 milliseconds. I use SimpleMessageBroker.

 public void publishToSessionUsingTopic(String sessionId, String subscriptionTopic, Map<String, CacheRowModel> payload) {

        String subscriptionTopicWithoutUser = subscriptionTopic.replace(USER_ENDPOINT, "");
        // necessary message headers for per-session send
        SimpMessageHeaderAccessor headerAccessor = SimpMessageHeaderAccessor.create(SimpMessageType.MESSAGE);
        headerAccessor.setSessionId(sessionId);
        headerAccessor.setLeaveMutable(true);          
        simpMessagingTemplate.convertAndSendToUser(sessionId, subscriptionTopicWithoutUser, Collections.singletonList(payload), headerAccessor.getMessageHeaders());

}

When this code is executed very frequently (every 2-3 milliseconds) for ~100 sessions, while I see in my logs that it was run and called the convertAndSendToUser, some of the sessions won't receive the message. I appreciate any suggestions about how this could be cleared.

Answer 1

Well, I think your problem is with the:

@Bean
public ThreadPoolTaskExecutor clientOutboundChannelExecutor() {
    TaskExecutorRegistration reg = getClientOutboundChannelRegistration().getOrCreateTaskExecRegistration();
    ThreadPoolTaskExecutor executor = reg.getTaskExecutor();
    executor.setThreadNamePrefix("clientOutboundChannel-");
    return executor;
}

where it uses this config for the Executor:

protected ThreadPoolTaskExecutor getTaskExecutor() {
    ThreadPoolTaskExecutor executor = (this.taskExecutor != null ? this.taskExecutor : new ThreadPoolTaskExecutor());
    executor.setCorePoolSize(this.corePoolSize);
    executor.setMaxPoolSize(this.maxPoolSize);
    executor.setKeepAliveSeconds(this.keepAliveSeconds);
    executor.setQueueCapacity(this.queueCapacity);
    executor.setAllowCoreThreadTimeOut(true);
    return executor;
}

See, there is no RejectedExecutionHandler configured. And by default it is like:

private RejectedExecutionHandler rejectedExecutionHandler = new ThreadPoolExecutor.AbortPolicy();

So, when you have enough many messages and tasks for them exceed the ThreadPool, any extra are just aborted.

To fix the issue you should implement WebSocketMessageBrokerConfigurer and override its configureClientOutboundChannel() to provide some custom taskExecutor(ThreadPoolTaskExecutor taskExecutor) for example with the new ThreadPoolExecutor.CallerRunsPolicy().