ServerSentEvents handler called again in case of client closes the connection

[yuryfunikov]

Bug report
edit: removed reactor-core:3.0.7.RELEASE dependency since it's not important for this issue
Using Spring Boot 1.5.7-RELEASE and Spring MVC and having just one application and controller classes:
The full example with test case is here https://github.com/yuryfunikov/spring-sse/tree/master

@SpringBootApplication//(exclude = {ErrorMvcAutoConfiguration.class})
public class SpringSseApplication {

	public static void main(String[] args) {
		SpringApplication.run(SpringSseApplication.class, args);
	}
}

@RestController
public class SseController {
private static final Logger logger = LoggerFactory.getLogger(SseController.class);

    @RequestMapping("/exception")
    public void exception() {
        throw new RuntimeException("test");
    }


   @RequestMapping("/sse")
    public SseEmitter sseEmitter2() {
        logger.info("/sse");
        SseEmitter emitter = new SseEmitter();

        new Thread(()->{
            for (int i = 0; i < 1000; i++) {
                try {
                    logger.info("next: {}", i);
                    emitter.send("next: " + i);
                } catch (IOException e) {
                    logger.info("IOException ");
                    break;
                }
                try {
                    Thread.sleep(1000L);
                } catch (InterruptedException e) {
                    logger.info("interrupted...");
                    break;
                }
            }
            emitter.complete();
        }).start();

        return emitter;

    }

}

Controller has request handler that returns SseEmitter object. Then i start SSE data consumer for instance using curl:
curl -i -H "Accept: application/json" http://localhost:8080/sse

In case if SSE data stream is stopped by client (e.g. with ctrl+c) there's a chance (~50%) that sseEmitter() method will be called again and you will see 'next: xxx' messages in server's console (though there's no connection to client anymore) without any errors. I.e. one will see console output like this:

2017-09-18 20:16:05.702  INFO 19416 --- [nio-8081-exec-3] com.example.springsse.SseController      : /sse
2017-09-18 20:16:05.703  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 0
2017-09-18 20:16:06.703  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 1
2017-09-18 20:16:07.707  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 2
2017-09-18 20:16:08.711  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 3
2017-09-18 20:16:09.714  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 4
2017-09-18 20:16:10.718  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 5
2017-09-18 20:16:11.722  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 6
2017-09-18 20:16:12.725  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : next: 7
2017-09-18 20:16:12.727  INFO 19416 --- [      Thread-14] o.apache.coyote.http11.Http11Processor   : An error occurred in processing while on a non-container thread. The connection will be closed immediately

java.io.IOException: An existing connection was forcibly closed by the remote host
	at sun.nio.ch.SocketDispatcher.write0(Native Method) ~[na:1.8.0_71]
	at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:51) ~[na:1.8.0_71]
	at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93) ~[na:1.8.0_71]
	at sun.nio.ch.IOUtil.write(IOUtil.java:65) ~[na:1.8.0_71]
	at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471) ~[na:1.8.0_71]
	at org.apache.tomcat.util.net.NioChannel.write(NioChannel.java:134) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.NioBlockingSelector.write(NioBlockingSelector.java:101) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.NioSelectorPool.write(NioSelectorPool.java:157) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.NioEndpoint$NioSocketWrapper.doWrite(NioEndpoint.java:1261) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.SocketWrapperBase.doWrite(SocketWrapperBase.java:670) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.SocketWrapperBase.flushBlocking(SocketWrapperBase.java:607) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.SocketWrapperBase.flush(SocketWrapperBase.java:597) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.http11.Http11OutputBuffer.flushBuffer(Http11OutputBuffer.java:581) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.http11.Http11OutputBuffer.flush(Http11OutputBuffer.java:272) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.http11.Http11Processor.flush(Http11Processor.java:1510) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.AbstractProcessor.action(AbstractProcessor.java:283) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.Response.action(Response.java:173) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.connector.OutputBuffer.doFlush(OutputBuffer.java:317) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.connector.OutputBuffer.flush(OutputBuffer.java:284) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.connector.CoyoteOutputStream.flush(CoyoteOutputStream.java:118) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at sun.nio.cs.StreamEncoder.implFlush(StreamEncoder.java:297) [na:1.8.0_71]
	at sun.nio.cs.StreamEncoder.flush(StreamEncoder.java:141) [na:1.8.0_71]
	at java.io.OutputStreamWriter.flush(OutputStreamWriter.java:229) [na:1.8.0_71]
	at org.springframework.util.StreamUtils.copy(StreamUtils.java:119) [spring-core-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.http.converter.StringHttpMessageConverter.writeInternal(StringHttpMessageConverter.java:106) [spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.http.converter.StringHttpMessageConverter.writeInternal(StringHttpMessageConverter.java:41) [spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.http.converter.AbstractHttpMessageConverter.write(AbstractHttpMessageConverter.java:227) [spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitterReturnValueHandler$HttpMessageConvertingHandler.sendInternal(ResponseBodyEmitterReturnValueHandler.java:207) [spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitterReturnValueHandler$HttpMessageConvertingHandler.send(ResponseBodyEmitterReturnValueHandler.java:200) [spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter.sendInternal(ResponseBodyEmitter.java:166) [spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter.send(ResponseBodyEmitter.java:159) [spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.SseEmitter.send(SseEmitter.java:126) [spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.SseEmitter.send(SseEmitter.java:107) [spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.SseEmitter.send(SseEmitter.java:89) [spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at com.example.springsse.SseController.lambda$sseEmitter2$0(SseController.java:32) [classes/:na]
	at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_71]

2017-09-18 20:16:12.733  INFO 19416 --- [      Thread-14] com.example.springsse.SseController      : IOException 
2017-09-18 20:16:12.737 ERROR 19416 --- [nio-8081-exec-4] o.a.c.c.C.[Tomcat].[localhost]           : Exception Processing ErrorPage[errorCode=0, location=/error]

org.apache.catalina.connector.ClientAbortException: java.io.IOException: An existing connection was forcibly closed by the remote host
	at org.apache.catalina.connector.OutputBuffer.doFlush(OutputBuffer.java:321) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.connector.OutputBuffer.flush(OutputBuffer.java:284) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.connector.CoyoteOutputStream.flush(CoyoteOutputStream.java:118) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at com.fasterxml.jackson.core.json.UTF8JsonGenerator.flush(UTF8JsonGenerator.java:1054) ~[jackson-core-2.8.10.jar:2.8.10]
	at com.fasterxml.jackson.databind.ObjectWriter.writeValue(ObjectWriter.java:953) ~[jackson-databind-2.8.10.jar:2.8.10]
	at org.springframework.http.converter.json.AbstractJackson2HttpMessageConverter.writeInternal(AbstractJackson2HttpMessageConverter.java:286) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.http.converter.AbstractGenericHttpMessageConverter.write(AbstractGenericHttpMessageConverter.java:106) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.AbstractMessageConverterMethodProcessor.writeWithMessageConverters(AbstractMessageConverterMethodProcessor.java:231) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.HttpEntityMethodProcessor.handleReturnValue(HttpEntityMethodProcessor.java:203) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.method.support.HandlerMethodReturnValueHandlerComposite.handleReturnValue(HandlerMethodReturnValueHandlerComposite.java:81) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:113) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:827) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:738) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:85) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.DispatcherServlet.doDispatch(DispatcherServlet.java:967) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.DispatcherServlet.doService(DispatcherServlet.java:901) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.FrameworkServlet.processRequest(FrameworkServlet.java:970) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.servlet.FrameworkServlet.doGet(FrameworkServlet.java:861) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at javax.servlet.http.HttpServlet.service(HttpServlet.java:635) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.springframework.web.servlet.FrameworkServlet.service(FrameworkServlet.java:846) ~[spring-webmvc-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at javax.servlet.http.HttpServlet.service(HttpServlet.java:742) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:231) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.springframework.web.filter.RequestContextFilter.doFilterInternal(RequestContextFilter.java:99) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:107) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:101) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:101) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.springframework.web.filter.OncePerRequestFilter.doFilter(OncePerRequestFilter.java:101) ~[spring-web-4.3.11.RELEASE.jar:4.3.11.RELEASE]
	at org.apache.catalina.core.ApplicationFilterChain.internalDoFilter(ApplicationFilterChain.java:193) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationFilterChain.doFilter(ApplicationFilterChain.java:166) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationDispatcher.invoke(ApplicationDispatcher.java:728) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationDispatcher.doInclude(ApplicationDispatcher.java:590) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.ApplicationDispatcher.include(ApplicationDispatcher.java:524) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.StandardHostValve.custom(StandardHostValve.java:389) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.StandardHostValve.status(StandardHostValve.java:254) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.StandardHostValve.throwable(StandardHostValve.java:349) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.core.AsyncContextImpl.setErrorState(AsyncContextImpl.java:422) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.connector.CoyoteAdapter.asyncDispatch(CoyoteAdapter.java:176) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.AbstractProcessor.dispatch(AbstractProcessor.java:225) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.AbstractProcessorLight.process(AbstractProcessorLight.java:53) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.AbstractProtocol$ConnectionHandler.process(AbstractProtocol.java:868) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.NioEndpoint$SocketProcessor.doRun(NioEndpoint.java:1457) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.SocketProcessorBase.run(SocketProcessorBase.java:49) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [na:1.8.0_71]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [na:1.8.0_71]
	at org.apache.tomcat.util.threads.TaskThread$WrappingRunnable.run(TaskThread.java:61) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at java.lang.Thread.run(Thread.java:745) [na:1.8.0_71]
Caused by: java.io.IOException: An existing connection was forcibly closed by the remote host
	at sun.nio.ch.SocketDispatcher.write0(Native Method) ~[na:1.8.0_71]
	at sun.nio.ch.SocketDispatcher.write(SocketDispatcher.java:51) ~[na:1.8.0_71]
	at sun.nio.ch.IOUtil.writeFromNativeBuffer(IOUtil.java:93) ~[na:1.8.0_71]
	at sun.nio.ch.IOUtil.write(IOUtil.java:65) ~[na:1.8.0_71]
	at sun.nio.ch.SocketChannelImpl.write(SocketChannelImpl.java:471) ~[na:1.8.0_71]
	at org.apache.tomcat.util.net.NioChannel.write(NioChannel.java:134) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.NioBlockingSelector.write(NioBlockingSelector.java:101) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.NioSelectorPool.write(NioSelectorPool.java:157) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.NioEndpoint$NioSocketWrapper.doWrite(NioEndpoint.java:1261) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.SocketWrapperBase.doWrite(SocketWrapperBase.java:670) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.SocketWrapperBase.flushBlocking(SocketWrapperBase.java:607) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.tomcat.util.net.SocketWrapperBase.flush(SocketWrapperBase.java:597) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.http11.Http11OutputBuffer.flushBuffer(Http11OutputBuffer.java:581) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.http11.Http11OutputBuffer.flush(Http11OutputBuffer.java:272) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.http11.Http11Processor.flush(Http11Processor.java:1510) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.AbstractProcessor.action(AbstractProcessor.java:283) [tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.coyote.Response.action(Response.java:173) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	at org.apache.catalina.connector.OutputBuffer.doFlush(OutputBuffer.java:317) ~[tomcat-embed-core-8.5.20.jar:8.5.20]
	... 52 common frames omitted

2017-09-18 20:16:12.739  INFO 19416 --- [nio-8081-exec-4] com.example.springsse.SseController      : /sse
2017-09-18 20:16:12.740  INFO 19416 --- [      Thread-15] com.example.springsse.SseController      : next: 0
2017-09-18 20:16:13.740  INFO 19416 --- [      Thread-15] com.example.springsse.SseController      : next: 1
2017-09-18 20:16:14.740  INFO 19416 --- [      Thread-15] com.example.springsse.SseController      : next: 2

The problem doesn't happen if you exclude ErrorMvcAutoConfiguration class.

[bclozel]

I must say this use case is very strange - why are you using SseEmitter and Flux? As of Spring 5, Spring MVC also supports Flux return types does the adapting work for you properly.

Note that using reactor, scheduling work on elastic and calling Thread.sleep in the middle of a reactive pipeline will lead to inconsistent, unsupported behavior.

Clients disconnecting abruptly will always be an issue and your server might write to the network before knowing the client has left. This can be reproduced in a real-world environment (e.g. with network latency) and your sample might just trigger that in an artificial manner.

Could we take a step back and focus on your use case and what you're trying to achieve?

[yuryfunikov]

Thank you for the answer! You are right that Flux is not relevant to the issue. I have updated my example and now it does't use reactor though the problem remains - method sseEmitter() is called twice for one execution of curl

[bclozel]

@yuryfunikov How are you running your application? In your IDE? As an executable JAR? Deployed a WAR in a container?

[yuryfunikov]

Hi,
from IDE (Idea) or from command line using smth like java -jar target/fat-jar. I have created a sample repository with example here https://github.com/yuryfunikov/spring-sse/tree/master you can use to see the problem. Bad news is that it seems ErrorMvcAutoConfiguration is not guilty and the issue happens with or without it.

[gopikori]

I am seeing exactly same issue in one of our projects. Thanks @yuryfunikov for providing an example code. That precisely shows my problem as well.

[bclozel]

So it seems your sample is doing the following:

• when the /sse endpoint is called, an SseEmitter is created to asynchronously write data to the response (in a non-container thread)
• this emitter periodically tries to write to the response
• the client closes the connection
• the emitter tries again to write to the connection, which does not exist anymore - an exception is thrown

Unlike your issue description, I don't see (with or without ErrorMvcAutoConfiguration) your sseEmitter endpoint being called again after the client disconnects (no /see INFO log).

Now as far a the spec goes, we are not notified when a client goes away, so trying to get the connection and write to it (thus, getting an exception) is the only way to know a client is gone.

I'm closing this issue as we can't address that limitation in Spring Boot or Spring Framework.
Feel free to reopen this issue if I've missed something.

Thanks!