diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java
index df98508fe..88b35de1d 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapter.java
@@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscription;
+
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DataBufferUtils;
@@ -74,11 +75,14 @@ class DataBufferPublisherAdapter {
 	@RequiredArgsConstructor
 	static class State {
 
-		private static final AtomicLongFieldUpdater<State> DEMAND = AtomicLongFieldUpdater.newUpdater(State.class, "demand");
+		private static final AtomicLongFieldUpdater<State> DEMAND = AtomicLongFieldUpdater.newUpdater(State.class,
+				"demand");
 
-		private static final AtomicIntegerFieldUpdater<State> STATE = AtomicIntegerFieldUpdater.newUpdater(State.class, "state");
+		private static final AtomicIntegerFieldUpdater<State> STATE = AtomicIntegerFieldUpdater.newUpdater(State.class,
+				"state");
 
-		private static final AtomicIntegerFieldUpdater<State> READ = AtomicIntegerFieldUpdater.newUpdater(State.class, "read");
+		private static final AtomicIntegerFieldUpdater<State> READ = AtomicIntegerFieldUpdater.newUpdater(State.class,
+				"read");
 
 		private static final int STATE_OPEN = 0;
 		private static final int STATE_CLOSED = 1;
@@ -123,6 +127,10 @@ class DataBufferPublisherAdapter {
 			return DEMAND.get(this);
 		}
 
+		boolean decrementDemand() {
+			return DEMAND.decrementAndGet(this) > 0;
+		}
+
 		void close() {
 			STATE.compareAndSet(this, STATE_OPEN, STATE_CLOSED);
 		}
@@ -141,7 +149,11 @@ class DataBufferPublisherAdapter {
 			DataBuffer dataBuffer = dataBufferFactory.allocateBuffer();
 			ByteBuffer intermediate = ByteBuffer.allocate(dataBuffer.capacity());
 
-			Mono.from(inputStream.read(intermediate)).subscribe(new BufferCoreSubscriber(sink, dataBuffer, intermediate));
+			try {
+				Mono.from(inputStream.read(intermediate)).subscribe(new BufferCoreSubscriber(sink, dataBuffer, intermediate));
+			} catch (Exception e) {
+				sink.error(e);
+			}
 		}
 
 		private class BufferCoreSubscriber implements CoreSubscriber<Integer> {
@@ -182,6 +194,7 @@ class DataBufferPublisherAdapter {
 				dataBuffer.write(intermediate);
 
 				sink.next(dataBuffer);
+				decrementDemand();
 
 				try {
 					if (bytes == -1) {
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapterUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapterUnitTests.java
new file mode 100644
index 000000000..62dc300a7
--- /dev/null
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/gridfs/DataBufferPublisherAdapterUnitTests.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright 2019 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.mongodb.gridfs;
+
+import static org.mockito.Mockito.*;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+
+import org.junit.Test;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.core.io.buffer.DefaultDataBufferFactory;
+
+/**
+ * Unit tests for {@link DataBufferPublisherAdapter}.
+ *
+ * @author Mark Paluch
+ */
+public class DataBufferPublisherAdapterUnitTests {
+
+	DataBufferFactory factory = new DefaultDataBufferFactory();
+
+	@Test // DATAMONGO-2230
+	public void adapterShouldPropagateErrors() {
+
+		AsyncInputStreamAdapter asyncInput = mock(AsyncInputStreamAdapter.class);
+
+		when(asyncInput.read(any())).thenReturn(Mono.just(1), Mono.error(new IllegalStateException()));
+		when(asyncInput.close()).thenReturn(Mono.empty());
+
+		Flux<DataBuffer> binaryStream = DataBufferPublisherAdapter.createBinaryStream(asyncInput, factory);
+
+		StepVerifier.create(binaryStream, 0) //
+				.thenRequest(1) //
+				.expectNextCount(1) //
+				.thenRequest(1) //
+				.verifyError(IllegalStateException.class);
+	}
+}