From 9277f1ce3c1e2d7711d3b3bed13c5f55e195c23a Mon Sep 17 00:00:00 2001 From: S O'Donnell Date: Thu, 4 Jun 2026 10:47:47 +0100 Subject: [PATCH 1/3] Correct potential race where the final items on the queue may not be read - corrupted read --- .../scm/storage/StreamBlockInputStream.java | 14 +- .../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 8a029f871719..dd3928f91378 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 9c77ae19f207..6275273c9f54 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 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 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 requestObserver = + mock(ClientCallStreamObserver.class); + StreamingReadResponse streamingReadResponse = mock(StreamingReadResponse.class); + when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver); + + AtomicReference 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(); + } } From 89eaef128b2082d2d54accd51d3c055864f7f9d0 Mon Sep 17 00:00:00 2001 From: S O'Donnell Date: Thu, 4 Jun 2026 11:15:41 +0100 Subject: [PATCH 2/3] Fix NPE / infinite loop when poll() returns null --- .../scm/storage/StreamBlockInputStream.java | 12 +- .../storage/TestStreamBlockInputStream.java | 131 ++++++++++++++++++ 2 files changed, 141 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 dd3928f91378..97bd81078251 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 @@ -451,17 +451,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 6275273c9f54..18b6e767b28e 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; @@ -258,6 +259,136 @@ public void testPollDoesNotDropQueuedItemWhenFutureCompletesFirst() throws Excep } } + /** + * 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 requestObserver = + mock(ClientCallStreamObserver.class); + StreamingReadResponse streamingReadResponse = mock(StreamingReadResponse.class); + when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver); + + AtomicReference 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 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 requestObserver = + mock(ClientCallStreamObserver.class); + StreamingReadResponse streamingReadResponse = mock(StreamingReadResponse.class); + when(streamingReadResponse.getRequestObserver()).thenReturn(requestObserver); + + AtomicReference 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 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. + assertDoesNotThrow(() -> sbis.read(buf), + "should not NPE when server completes stream without covering the seek position"); + } + } + private OzoneClientConfig newStreamReadConfig() { OzoneClientConfig clientConfig = new OzoneClientConfig(); clientConfig.setChecksumVerify(false); From d80a640fb25b609ae0f2489ce7c60d19b05dc3ac Mon Sep 17 00:00:00 2001 From: Siyao Meng <50227127+smengcl@users.noreply.github.com> Date: Thu, 9 Jul 2026 14:23:40 -0700 Subject: [PATCH 3/3] Update hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestStreamBlockInputStream.java --- .../hadoop/hdds/scm/storage/TestStreamBlockInputStream.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 46688b3c6367..4483945509bb 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 @@ -435,8 +435,9 @@ public void testReadFromQueueNpeWhenStreamCompletesWithoutCoveringSeekPosition() 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. - assertDoesNotThrow(() -> sbis.read(buf), - "should not NPE when server completes stream without covering the seek position"); + 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"); } }