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 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(), 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(), 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