This is an automated email from the ASF dual-hosted git repository.
smengcl 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 a0f08cc3f0b HDDS-15480. Potential for NPE / infinite loop in
StreamBlockReader (#10431)
a0f08cc3f0b is described below
commit a0f08cc3f0bd0f4d2d7625e8d83d0c81b340c263
Author: Stephen O'Donnell <[email protected]>
AuthorDate: Thu Jul 9 23:56:59 2026 +0100
HDDS-15480. Potential for NPE / infinite loop in StreamBlockReader (#10431)
Co-authored-by: S O'Donnell <[email protected]>
Co-authored-by: Siyao Meng <[email protected]>
---
.../hdds/scm/storage/StreamBlockInputStream.java | 12 +-
.../scm/storage/TestStreamBlockInputStream.java | 132 +++++++++++++++++++++
2 files changed, 142 insertions(+), 2 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 975c0511832..b008abdb56c 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
@@ -482,17 +482,25 @@ private ByteBuffer read(int length, boolean preRead)
throws IOException {
while (true) {
final ByteBuffer buf = readFromQueue();
- if (buf != null && buf.hasRemaining()) {
+ if (buf == null) {
+ return null; // Stream ended
+ }
+ if (buf.hasRemaining()) {
return buf;
}
+ // buf is empty: the server aligned its response to a checksum boundary
+ // before our current position and all bytes were skipped. Fetch the
next
+ // response, which should start at or after our position.
}
}
ByteBuffer readFromQueue() throws IOException {
final ReadBlockResponseProto readBlock = poll();
+ if (readBlock == null) {
+ return null; // Stream ended
+ }
// The server always returns data starting from the last checksum
boundary. Therefore if the reader position is
// ahead of the position we received from the server, we need to adjust
the buffer position accordingly.
- // If the reader position is behind
final ByteString data = readBlock.getData();
final ByteBuffer dataBuffer = data.asReadOnlyByteBuffer();
final long blockOffset = readBlock.getOffset();
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 9e1ff0893ec..4483945509b 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.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
@@ -309,6 +310,137 @@ private XceiverClientGrpc mockStreamingReadClient(byte[]
data,
return xceiverClient;
}
+ /**
+ * Realistic test for the checksum-alignment skip path.
+ *
+ * After a seek, the server aligns its response to the nearest checksum
boundary,
+ * which may be before the client's current position. With a small
responseDataSize
+ * (4 bytes), the server sends two 4-byte chunks:
+ * chunk 1: blockOffset=0, data=[0,1,2,3] — entirely before seek position 4
+ * chunk 2: blockOffset=4, data=[4,5,6,7] — starts at seek position
+ *
+ * The while(true) loop in read() must:
+ * iteration 1: receive chunk 1, skip all 4 bytes (pos-blockOffset=4 ==
data.size()),
+ * empty buffer → continue
+ * iteration 2: receive chunk 2, no skip needed → return buffer with
[4,5,6,7]
+ *
+ * This was an infinite loop or MPE before fixes in the PR that added this
test.
+ */
+ @Test
+ public void
testSeekReadsCorrectBytesWhenFirstResponseIsFullyBeforePosition() throws
Exception {
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ clientConfig.setStreamReadResponseDataSize(4); // 4-byte chunks match the
test data
+ BlockID blockID = new BlockID(1L, 12L);
+ long length = 8;
+ 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(), any());
+
+ // Server aligns to checksum boundary 0 and sends two 4-byte responses.
+ // The first chunk (bytes 0–3) is entirely before seek position 4 and will
be
+ // fully skipped. The second chunk (bytes 4–7) starts at our position.
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = readerRef.get();
+ reader.onNext(buildResponseProto(new byte[]{0, 1, 2, 3}, 0)); // fully
skipped
+ reader.onNext(buildResponseProto(new byte[]{4, 5, 6, 7}, 4)); // has our
data
+ reader.onCompleted();
+ 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)) {
+
+ sbis.seek(4);
+
+ byte[] out = new byte[4];
+ int bytesRead = sbis.read(out, 0, 4);
+ assertEquals(4, bytesRead);
+ assertArrayEquals(new byte[]{4, 5, 6, 7}, out,
+ "should return bytes starting from seek position, skipping the
checksum-aligned preamble");
+ }
+ }
+
+ /**
+ * Defensive test for readFromQueue() which NPE'ed when poll() returns null.
+ *
+ * This tests a server-error / edge-case scenario: the server sends only a
+ * single response whose data ends before the client's seek position, then
+ * immediately completes the stream. The while(true) loop in read() skips
+ * all bytes in the response (empty buffer), then calls poll() again. poll()
+ * finds the queue empty and isDone()==true and returns null.
+ *
+ * A well-behaved server would never complete the stream without covering the
+ * client's position, so this scenario represents a protocol violation rather
+ * than normal operation, but it serves to reproduce the NPE exception before
+ * fixing the code.
+ */
+ @Test
+ public void
testReadFromQueueNpeWhenStreamCompletesWithoutCoveringSeekPosition() throws
Exception {
+ OzoneClientConfig clientConfig = newStreamReadConfig();
+ // Short timeout so the test completes quickly rather than waiting 5 s.
+ clientConfig.setStreamReadTimeout(Duration.ofMillis(200));
+
+ BlockID blockID = new BlockID(1L, 13L);
+ long length = 8;
+ 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(), any());
+
+ // Server only sends bytes 0–3 (before seek position 4) then completes —
+ // simulating a protocol violation or a truncated/corrupt response.
+ doAnswer(inv -> {
+ StreamingReaderSpi reader = readerRef.get();
+ reader.onNext(buildResponseProto(new byte[]{0, 1, 2, 3}, 0));
+ reader.onCompleted();
+ 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)) {
+
+ sbis.seek(4);
+
+ ByteBuffer buf = ByteBuffer.allocate(4);
+ // Before the fixes: threw NullPointerException (Bug 1) or looped
forever (Bug 3).
+ // After the fixes: returns gracefully with 0 / EOF rather than crashing.
+ int bytesRead = sbis.read(buf);
+ assertEquals(-1, bytesRead, "should reach EOF when the stream completes
before the seek position");
+ assertEquals(0, buf.position(), "no bytes should be produced");
+ }
+ }
+
/**
* When the server delivers multiple responses plus onCompleted() inside a
* single streamRead() call (all on the same call stack), the first response
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
