diff --git a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java
index e048576f96..198bbfa024 100644
--- a/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java
+++ b/spring-core/src/main/java/org/springframework/core/io/buffer/DataBufferUtils.java
@@ -139,12 +139,14 @@ public abstract class DataBufferUtils {
 		DataBuffer dataBuffer = dataBufferFactory.allocateBuffer(bufferSize);
 		ByteBuffer byteBuffer = dataBuffer.asByteBuffer(0, bufferSize);
 
+
 		Flux<DataBuffer> result = Flux.using(channelSupplier,
 				channel -> Flux.create(sink -> {
-					CompletionHandler<Integer, DataBuffer> completionHandler =
+					AsynchronousFileChannelReadCompletionHandler completionHandler =
 							new AsynchronousFileChannelReadCompletionHandler(channel,
 									sink, position, dataBufferFactory, bufferSize);
 					channel.read(byteBuffer, position, dataBuffer, completionHandler);
+					sink.onDispose(completionHandler::dispose);
 				}),
 				DataBufferUtils::closeChannel);
 
@@ -545,6 +547,10 @@ public abstract class DataBufferUtils {
 			release(dataBuffer);
 			this.sink.error(exc);
 		}
+
+		public void dispose() {
+			this.disposed.set(true);
+		}
 	}
 
 
diff --git a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java
index 8f53d733fe..ff511291db 100644
--- a/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java
+++ b/spring-core/src/test/java/org/springframework/core/io/buffer/DataBufferUtilsTests.java
@@ -40,6 +40,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import org.springframework.core.io.ByteArrayResource;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.support.DataBufferTestUtils;
@@ -142,6 +143,36 @@ public class DataBufferUtilsTests extends AbstractDataBufferAllocatingTestCase {
 				.verify(Duration.ofSeconds(5));
 	}
 
+	@Test
+	public void readResourcePositionAndTakeUntil() throws Exception {
+		Resource resource = new ClassPathResource("DataBufferUtilsTests.txt", getClass());
+		Flux<DataBuffer> flux = DataBufferUtils.read(resource, 3, this.bufferFactory, 3);
+
+		flux = DataBufferUtils.takeUntilByteCount(flux, 5);
+
+
+		StepVerifier.create(flux)
+				.consumeNextWith(stringConsumer("bar"))
+				.consumeNextWith(stringConsumer("ba"))
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
+	}
+
+	@Test
+	public void readByteArrayResourcePositionAndTakeUntil() throws Exception {
+		Resource resource = new ByteArrayResource("foobarbazqux" .getBytes());
+		Flux<DataBuffer> flux = DataBufferUtils.read(resource, 3, this.bufferFactory, 3);
+
+		flux = DataBufferUtils.takeUntilByteCount(flux, 5);
+
+
+		StepVerifier.create(flux)
+				.consumeNextWith(stringConsumer("bar"))
+				.consumeNextWith(stringConsumer("ba"))
+				.expectComplete()
+				.verify(Duration.ofSeconds(5));
+	}
+
 	@Test
 	public void writeOutputStream() throws Exception {
 		DataBuffer foo = stringBuffer("foo");