Ensure ISE will not be thrown when the client disconnects

- ServletServerHttpResponse.ResponseAsyncListener#onError/onTimeout
must complete the async operation
- ServletHttpHandlerAdapter.HandlerResultSubscriber#onComplete must
check that the async operation is not completed

Issue: SPR-15412

spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java

@@ -49,7 +49,6 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 
 	private final AtomicReference<State> state = new AtomicReference<>(State.UNSUBSCRIBED);
 
-	@SuppressWarnings("unused")
 	private volatile long demand;
 
 	@SuppressWarnings("rawtypes")
@@ -96,8 +95,8 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 	 * Listeners can call this to notify when a read error has occurred.
 	 */
 	public final void onError(Throwable t) {
-		if (this.logger.isErrorEnabled()) {
-			this.logger.error(this.state + " onError: " + t, t);
+		if (this.logger.isTraceEnabled()) {
+			this.logger.trace(this.state + " onError: " + t);
 		}
 		this.state.get().onError(this, t);
 	}

spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java

@@ -166,16 +166,32 @@ public class ServletHttpHandlerAdapter implements Servlet {
 
 		@Override
 		public void onError(Throwable ex) {
-			logger.error("Could not complete request", ex);
-			HttpServletResponse response = (HttpServletResponse) this.asyncContext.getResponse();
-			response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
-			this.asyncContext.complete();
+			ServletRequest request = getRequest();
+			if (request != null && request.isAsyncStarted()) {
+				logger.error("Could not complete request", ex);
+				HttpServletResponse response = (HttpServletResponse) this.asyncContext.getResponse();
+				response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
+				this.asyncContext.complete();
+			}
 		}
 
 		@Override
 		public void onComplete() {
-			logger.debug("Successfully completed request");
-			this.asyncContext.complete();
+			ServletRequest request = getRequest();
+			if (request != null && request.isAsyncStarted()) {
+				logger.debug("Successfully completed request");
+				this.asyncContext.complete();
+			}
+		}
+
+		private ServletRequest getRequest() {
+			ServletRequest request = null;
+			try {
+				request = this.asyncContext.getRequest();
+			} catch (IllegalStateException ignore) {
+				// AsyncContext has been recycled and should not be used
+			}
+			return request;
 		}
 	}
 

spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java

@@ -180,11 +180,13 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons
 			Throwable ex = event.getThrowable();
 			ex = (ex != null ? ex : new IllegalStateException("Async operation timeout."));
 			handleError(ex);
+			event.getAsyncContext().complete();
 		}
 
 		@Override
 		public void onError(AsyncEvent event) {
 			handleError(event.getThrowable());
+			event.getAsyncContext().complete();
 		}
 
 		void handleError(Throwable ex) {