diff --git a/spring-core/src/main/java/org/springframework/util/StreamUtils.java b/spring-core/src/main/java/org/springframework/util/StreamUtils.java
index a64d1d0a7ae..fb51c710290 100644
--- a/spring-core/src/main/java/org/springframework/util/StreamUtils.java
+++ b/spring-core/src/main/java/org/springframework/util/StreamUtils.java
@@ -221,9 +221,13 @@ public abstract class StreamUtils {
 	 * Return a variant of the given {@link InputStream} where calling
 	 * {@link InputStream#close() close()} has no effect.
 	 * @param in the InputStream to decorate
-	 * @return a version of the InputStream that ignores calls to close
+	 * @return a version of the InputStream that ignores calls to close,
+	 * or the given InputStream if it is non-closing already
 	 */
 	public static InputStream nonClosing(InputStream in) {
+		if (in instanceof NonClosingInputStream) {
+			return in;
+		}
 		Assert.notNull(in, "No InputStream specified");
 		return new NonClosingInputStream(in);
 	}
@@ -232,15 +236,38 @@ public abstract class StreamUtils {
 	 * Return a variant of the given {@link OutputStream} where calling
 	 * {@link OutputStream#close() close()} has no effect.
 	 * @param out the OutputStream to decorate
-	 * @return a version of the OutputStream that ignores calls to close
+	 * @return a version of the OutputStream that ignores calls to close,
+	 * or the given OutputStream if it is non-closing already
+	 * @see #nonFlushing(OutputStream)
 	 */
 	public static OutputStream nonClosing(OutputStream out) {
+		if (out instanceof NonClosingOutputStream) {
+			return out;
+		}
 		Assert.notNull(out, "No OutputStream specified");
 		return new NonClosingOutputStream(out);
 	}
 
+	/**
+	 * Return a variant of the given {@link OutputStream} where calling
+	 * {@link OutputStream#flush() flush()} and/or
+	 * {@link OutputStream#close() close()} has no effect.
+	 * @param out the OutputStream to decorate
+	 * @return a version of the OutputStream that ignores calls to flush/close,
+	 * or the given OutputStream if it is non-flushing already
+	 * @since 7.0.6
+	 * @see #nonClosing(OutputStream)
+	 */
+	public static OutputStream nonFlushing(OutputStream out) {
+		if (out instanceof NonFlushingOutputStream) {
+			return out;
+		}
+		Assert.notNull(out, "No OutputStream specified");
+		return new NonFlushingOutputStream(out);
+	}
+
 
-	private static final class NonClosingInputStream extends FilterInputStream {
+	private static class NonClosingInputStream extends FilterInputStream {
 
 		public NonClosingInputStream(InputStream in) {
 			super(in);
@@ -272,7 +299,7 @@ public abstract class StreamUtils {
 	}
 
 
-	private static final class NonClosingOutputStream extends FilterOutputStream {
+	private static class NonClosingOutputStream extends FilterOutputStream {
 
 		public NonClosingOutputStream(OutputStream out) {
 			super(out);
@@ -289,4 +316,16 @@ public abstract class StreamUtils {
 		}
 	}
 
+
+	private static class NonFlushingOutputStream extends NonClosingOutputStream {
+
+		public NonFlushingOutputStream(OutputStream out) {
+			super(out);
+		}
+
+		@Override
+		public void flush() throws IOException {
+		}
+	}
+
 }
diff --git a/spring-core/src/test/java/org/springframework/util/StreamUtilsTests.java b/spring-core/src/test/java/org/springframework/util/StreamUtilsTests.java
index 974c9da0d54..ad53caa4e7a 100644
--- a/spring-core/src/test/java/org/springframework/util/StreamUtilsTests.java
+++ b/spring-core/src/test/java/org/springframework/util/StreamUtilsTests.java
@@ -124,6 +124,7 @@ class StreamUtilsTests {
 	void nonClosingInputStream() throws Exception {
 		InputStream source = mock();
 		InputStream nonClosing = StreamUtils.nonClosing(source);
+
 		nonClosing.read();
 		nonClosing.read(bytes);
 		nonClosing.read(bytes, 1, 2);
@@ -133,21 +134,49 @@ class StreamUtilsTests {
 		ordered.verify(source).read(bytes, 0, bytes.length);
 		ordered.verify(source).read(bytes, 1, 2);
 		ordered.verify(source, never()).close();
+
+		assertThat(StreamUtils.nonClosing(nonClosing)).isSameAs(nonClosing);
 	}
 
 	@Test
 	void nonClosingOutputStream() throws Exception {
 		OutputStream source = mock();
 		OutputStream nonClosing = StreamUtils.nonClosing(source);
+
 		nonClosing.write(1);
 		nonClosing.write(bytes);
 		nonClosing.write(bytes, 1, 2);
+		nonClosing.flush();
 		nonClosing.close();
 		InOrder ordered = inOrder(source);
 		ordered.verify(source).write(1);
 		ordered.verify(source).write(bytes, 0, bytes.length);
 		ordered.verify(source).write(bytes, 1, 2);
+		ordered.verify(source).flush();
 		ordered.verify(source, never()).close();
+
+		assertThat(StreamUtils.nonClosing(nonClosing)).isSameAs(nonClosing);
+	}
+
+	@Test
+	void nonFlushingOutputStream() throws Exception {
+		OutputStream source = mock();
+		OutputStream nonFlushing = StreamUtils.nonFlushing(source);
+
+		nonFlushing.write(1);
+		nonFlushing.write(bytes);
+		nonFlushing.write(bytes, 1, 2);
+		nonFlushing.flush();
+		nonFlushing.close();
+		InOrder ordered = inOrder(source);
+		ordered.verify(source).write(1);
+		ordered.verify(source).write(bytes, 0, bytes.length);
+		ordered.verify(source).write(bytes, 1, 2);
+		ordered.verify(source, never()).flush();
+		ordered.verify(source, never()).close();
+
+		assertThat(StreamUtils.nonFlushing(nonFlushing)).isSameAs(nonFlushing);
+		assertThat(StreamUtils.nonClosing(nonFlushing)).isSameAs(nonFlushing);
 	}
 
 }