This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 865d5ac5261 HDDS-15479. Possible incomplete read as
StreamBlockInputStream may drop last queued item (#10430)
865d5ac5261 is described below
commit 865d5ac526130af038359ffe96b4934157d7f939
Author: Stephen O'Donnell <[email protected]>
AuthorDate: Thu Jun 11 14:03:17 2026 +0100
HDDS-15479. Possible incomplete read as StreamBlockInputStream may drop
last queued item (#10430)
---
.../hdds/scm/storage/StreamBlockInputStream.java | 14 +-
.../scm/storage/TestStreamBlockInputStream.java | 145 +++++++++++++++++++++
2 files changed, 155 insertions(+), 4 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
index 8a029f87171..dd3928f9137 100644
---
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
+++
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java
@@ -411,9 +411,6 @@ ReadBlockResponseProto poll() throws IOException {
while (true) {
checkError();
- if (future.isDone()) {
- return null; // Stream ended
- }
final ReadBlockResponseProto proto;
try {
@@ -426,6 +423,13 @@ ReadBlockResponseProto poll() throws IOException {
return proto;
}
+ // Check isDone only after confirming the queue is empty. If isDone()
were
+ // checked first, an item delivered by onNext() just before
onCompleted()
+ // fired would be silently dropped, causing data corruption.
+ if (future.isDone()) {
+ return null; // Stream ended, queue is empty
+ }
+
final long elapsedNanos = System.nanoTime() - startTime;
if (elapsedNanos >= readTimeoutNanos) {
setFailedAndThrow(new TimeoutIOException(
@@ -438,7 +442,9 @@ ReadBlockResponseProto poll() throws IOException {
private ByteBuffer read(int length, boolean preRead) throws IOException {
checkError();
if (future.isDone()) {
- return null; // Stream ended
+ // Don't return null while items remain in the queue. onNext() may
have delivered items just before
+ // onCompleted() fired.
+ return responseQueue.isEmpty() ? null : readFromQueue();
}
readBlock(length, preRead);
diff --git
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
index 9c77ae19f20..6275273c9f5 100644
---
a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
+++
b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java
@@ -17,6 +17,7 @@
package org.apache.hadoop.hdds.scm.storage;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
@@ -32,6 +33,7 @@
import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.apache.hadoop.hdds.client.BlockID;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -188,6 +190,74 @@ public void
testCloseDoesNotFailWhenOnCompletedAndCancelThrow() throws Exception
verify(xceiverClient, times(1)).completeStreamRead();
}
+ /**
+ * Reproduces Bug 2: poll() checks future.isDone() before draining the queue.
+ *
+ * When the server delivers a response (onNext) and immediately closes the
stream
+ * (onCompleted) — which can happen on the same gRPC thread in rapid
succession —
+ * the item is in the queue and the future is already complete by the time
poll()
+ * first runs. poll() sees isDone()==true and returns null without ever
checking
+ * the queue, so readFromQueue() throws NullPointerException on the null
proto.
+ *
+ * This test will FAIL with NullPointerException on the current code and
should
+ * PASS once the bug is fixed (poll must drain the queue before checking
isDone).
+ */
+ @Test
+ public void testPollDoesNotDropQueuedItemWhenFutureCompletesFirst() throws
Exception {
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ BlockID blockID = new BlockID(1L, 10L);
+ byte[] data = {1, 2, 3, 4};
+ Pipeline pipeline = mockStandalonePipeline();
+ ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver =
+ mock(ClientCallStreamObserver.class);
+ StreamingReadResponse streamingReadResponse =
mock(StreamingReadResponse.class);
+
when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver);
+
+ // Capture the StreamingReaderSpi during initStreamRead so we can drive
+ // its callbacks from the streamRead mock below.
+ AtomicReference<StreamingReaderSpi> readerRef = new AtomicReference<>();
+ XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = inv.getArgument(1);
+ reader.setStreamingReadResponse(streamingReadResponse);
+ readerRef.set(reader);
+ return null;
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+
+ // Simulate the race: when the client sends a ReadBlock request, the server
+ // responds with data (onNext) and closes the stream (onCompleted) before
+ // poll() has had a chance to run — both callbacks fire on the same call
stack
+ // before streamRead() returns. This means when poll() is entered, the
queue
+ // already has the response item AND future.isDone() is already true.
+ // poll() checks isDone() first and returns null, dropping the queued item.
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = readerRef.get();
+ reader.onNext(ContainerCommandResponseProto.newBuilder()
+ .setCmdType(Type.ReadBlock)
+ .setResult(ContainerProtos.Result.SUCCESS)
+ .setReadBlock(buildReadBlockResponse(data))
+ .build());
+ reader.onCompleted(); // future is now done; item is already in the queue
+ return null;
+ }).when(xceiverClient).streamRead(any(), any());
+
+ XceiverClientFactory xceiverClientFactory =
mock(XceiverClientFactory.class);
+ when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class)))
+ .thenReturn(xceiverClient);
+
+ try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+ blockID, data.length, pipeline, null, xceiverClientFactory,
+ NO_REFRESH, clientConfig)) {
+
+ ByteBuffer buf = ByteBuffer.allocate(data.length);
+ // With the bug: poll() returns null (future done, queue unchecked) and
+ // readFromQueue() throws NullPointerException.
+ // After the fix: all 4 bytes are returned successfully.
+ assertDoesNotThrow(() -> sbis.read(buf), "should not NPE when
onCompleted fires before poll");
+ assertEquals(data.length, buf.position(), "all bytes should be read");
+ }
+ }
+
private OzoneClientConfig newStreamReadConfig() {
OzoneClientConfig clientConfig = new OzoneClientConfig();
clientConfig.setChecksumVerify(false);
@@ -237,6 +307,66 @@ private XceiverClientGrpc mockStreamingReadClient(byte[]
data,
return xceiverClient;
}
+ /**
+ * When the server delivers multiple responses plus onCompleted() inside a
+ * single streamRead() call (all on the same call stack), the first response
+ * is consumed correctly, but by the time read() is invoked again for the
+ * second chunk, future.isDone() is already true. read() sees isDone() and
+ * returns null immediately without checking the queue, so the second (and
+ * any further) queued responses are silently dropped.
+ */
+ @Test
+ public void testReadDoesNotDropQueuedItemsWhenFutureIsDoneOnSecondCall()
throws Exception {
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ BlockID blockID = new BlockID(1L, 11L);
+ byte[] firstChunk = {1, 2, 3, 4};
+ byte[] secondChunk = {5, 6, 7, 8};
+ long length = firstChunk.length + secondChunk.length; // 8 bytes total
+
+ Pipeline pipeline = mockStandalonePipeline();
+ ClientCallStreamObserver<ContainerCommandRequestProto> requestObserver =
+ mock(ClientCallStreamObserver.class);
+ StreamingReadResponse streamingReadResponse =
mock(StreamingReadResponse.class);
+
when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver);
+
+ AtomicReference<StreamingReaderSpi> readerRef = new AtomicReference<>();
+ XceiverClientGrpc xceiverClient = mock(XceiverClientGrpc.class);
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = inv.getArgument(1);
+ reader.setStreamingReadResponse(streamingReadResponse);
+ readerRef.set(reader);
+ return null;
+ }).when(xceiverClient).initStreamRead(any(BlockID.class), any());
+
+ // Server delivers both 4-byte chunks plus onCompleted() in one synchronous
+ // call. After streamRead() returns: queue=[chunk1, chunk2], isDone=true.
+ // read() correctly returns chunk1 on the first call, but on the second
call
+ // it sees isDone()==true and returns null before draining chunk2.
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = readerRef.get();
+ reader.onNext(buildResponseProto(firstChunk, 0));
+ reader.onNext(buildResponseProto(secondChunk, firstChunk.length));
+ reader.onCompleted(); // future done; both items still in queue
+ return null;
+ }).when(xceiverClient).streamRead(any(), any());
+
+ XceiverClientFactory xceiverClientFactory =
mock(XceiverClientFactory.class);
+ when(xceiverClientFactory.acquireClientForReadData(any(Pipeline.class)))
+ .thenReturn(xceiverClient);
+
+ try (StreamBlockInputStream sbis = new StreamBlockInputStream(
+ blockID, length, pipeline, null, xceiverClientFactory,
+ NO_REFRESH, clientConfig)) {
+ ByteBuffer buf = ByteBuffer.allocate((int) length);
+ // With the bug: read() returns null on the second call (isDone is true),
+ // so only 4 bytes are read and buf.position() == 4.
+ // After the fix: all 8 bytes are read and buf.position() == 8.
+ int bytesRead = sbis.read(buf);
+ assertEquals(length, bytesRead, "expected all bytes to be read");
+ assertEquals(length, buf.position(), "buffer position should be at end
of block");
+ }
+ }
+
private ReadBlockResponseProto buildReadBlockResponse(byte[] data) {
return ReadBlockResponseProto.newBuilder()
.setOffset(0)
@@ -247,4 +377,19 @@ private ReadBlockResponseProto
buildReadBlockResponse(byte[] data) {
.build())
.build();
}
+
+ private ContainerCommandResponseProto buildResponseProto(byte[] data, long
offset) {
+ return ContainerCommandResponseProto.newBuilder()
+ .setCmdType(Type.ReadBlock)
+ .setResult(ContainerProtos.Result.SUCCESS)
+ .setReadBlock(ReadBlockResponseProto.newBuilder()
+ .setOffset(offset)
+ .setData(ByteString.copyFrom(data))
+ .setChecksumData(ChecksumData.newBuilder()
+ .setType(ContainerProtos.ChecksumType.NONE)
+ .setBytesPerChecksum(data.length)
+ .build())
+ .build())
+ .build();
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
