diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java
index 71adf24dd16329a0822cf562d48fef6e1e114509..811cb78a96c80725956e4c3ce4c37a678eb3cb7d 100644
--- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java
+++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java
@@ -106,6 +106,10 @@ abstract class AbstractResponseBodyFlushProcessor
 	 */
 	protected abstract void flush() throws IOException;
 
+	private void cancel() {
+		this.subscription.cancel();
+	}
+
 	private void writeComplete() {
 		if (logger.isTraceEnabled()) {
 			logger.trace(this.state + " writeComplete");
@@ -157,11 +161,12 @@ abstract class AbstractResponseBodyFlushProcessor
 				else {
 					try {
 						processor.flush();
+						processor.subscription.request(1);
 					}
 					catch (IOException ex) {
+						processor.cancel();
 						processor.onError(ex);
 					}
-					processor.subscription.request(1);
 				}
 			}
 		}, COMPLETED {
@@ -231,6 +236,7 @@ abstract class AbstractResponseBodyFlushProcessor
 
 			@Override
 			public void onError(Throwable t) {
+				processor.cancel();
 				processor.onError(t);
 			}
 
diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java
index 1a0268cd07d80b0a0df1f82d23847d84c2f1821b..2463fb941438b74b565a368d8caf61621405b759 100644
--- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java
+++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java
@@ -159,6 +159,10 @@ abstract class AbstractResponseBodyProcessor implements Processor<DataBuffer, Vo
 	 */
 	protected abstract boolean write(DataBuffer dataBuffer) throws IOException;
 
+	protected void cancel() {
+		this.subscription.cancel();
+	}
+
 	private boolean changeState(State oldState, State newState) {
 		return this.state.compareAndSet(oldState, newState);
 	}
@@ -220,7 +224,6 @@ abstract class AbstractResponseBodyProcessor implements Processor<DataBuffer, Vo
 			@Override
 			void onComplete(AbstractResponseBodyProcessor processor) {
 				if (processor.changeState(this, COMPLETED)) {
-					processor.subscriberCompleted = true;
 					processor.publisherDelegate.publishComplete();
 				}
 			}
@@ -258,6 +261,7 @@ abstract class AbstractResponseBodyProcessor implements Processor<DataBuffer, Vo
 						}
 					}
 					catch (IOException ex) {
+						processor.cancel();
 						processor.onError(ex);
 					}
 				}
diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java
index 323f8c0a868b25807dfb15f80fb62727bc6f9ef0..c1e13d2a85b269d353479b451467f8aef99da808 100644
--- a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java
+++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java
@@ -224,6 +224,7 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons
 		@Override
 		public void onError(Throwable ex) {
 			if (bodyProcessor != null) {
+				bodyProcessor.cancel();
 				bodyProcessor.onError(ex);
 			}
 		}