From 3479dd820c0dfc151b2b7e7901b32e60b0e06160 Mon Sep 17 00:00:00 2001 From: Tsz-Wo Nicholas Sze Date: Mon, 1 Jun 2026 22:01:49 -0700 Subject: [PATCH 1/7] HDDS-15422. Stream read seek should not close stream --- .../scm/storage/StreamBlockInputStream.java | 39 +++++++++++-------- 1 file changed, 22 insertions(+), 17 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..e347e33e98d2 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 @@ -110,6 +110,8 @@ public StreamBlockInputStream( this.responseDataSize = config.getStreamReadResponseDataSize(); this.readTimeout = config.getStreamReadTimeout(); this.readTimeoutNanos = readTimeout.toNanos(); + + LOG.debug("{}: new StreamBlockInputStream", name); } @Override @@ -130,11 +132,12 @@ public synchronized long getPos() { @Override public synchronized int read() throws IOException { checkOpen(); - if (!dataAvailableToRead(1, true)) { + final boolean preRead = true; + if (!dataAvailableToRead(1, preRead)) { return EOF; } int value = buffer.get(); - advancePosition(1); + advancePosition(1, preRead); return value; } @@ -161,7 +164,7 @@ synchronized int readFully(ByteBuffer targetBuf, boolean preRead) throws IOExcep tmpBuf.limit(tmpBuf.position() + toCopy); targetBuf.put(tmpBuf); buffer.position(tmpBuf.position()); - advancePosition(toCopy); + advancePosition(toCopy, preRead); read += toCopy; } return read > 0 ? read : EOF; @@ -180,10 +183,11 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th return bufferHasRemaining(); } - private synchronized void advancePosition(long delta) { + private synchronized void advancePosition(long delta, boolean preRead) { + LOG.debug("{}: advance {} -> {}", getName(streamingReader), position, position + delta); position += delta; - if (position >= blockLength && streamingReader != null) { - closeStream(); + if (preRead && position >= blockLength) { + closeReader("advancePosition"); } } @@ -208,8 +212,8 @@ public synchronized void seek(long pos) throws IOException { if (pos == position) { return; } - LOG.debug("{}: seek {} -> {}", this, position, pos); - closeStream(); + LOG.debug("{}: seek {} -> {}", getName(streamingReader), position, pos); + buffer = null; position = pos; requestedLength = pos; } @@ -226,7 +230,7 @@ public synchronized void unbuffer() { releaseClient(); } - private synchronized void closeStream() { + private synchronized void closeReader(String reason) { if (streamingReader == null) { buffer = null; return; @@ -235,10 +239,7 @@ private synchronized void closeStream() { final StreamingReader reader = streamingReader; streamingReader = null; buffer = null; - - if (LOG.isDebugEnabled()) { - LOG.debug("Closing {}", reader); - } + LOG.debug("{} closeReader for {}", getName(reader), reason); reader.onCompleted(); @@ -293,6 +294,7 @@ private synchronized void initialize() throws IOException { try { acquireClient(); final StreamingReader reader = new StreamingReader(); + LOG.debug("{}: new StreamingReader", getName(reader)); xceiverClient.initStreamRead(blockID, reader); streamingReader = reader; } catch (IOException ioe) { @@ -342,7 +344,7 @@ private void handleExceptions(IOException cause) throws IOException { protected synchronized void releaseClient() { if (xceiverClientFactory != null && xceiverClient != null) { - closeStream(); + closeReader("releaseClient"); xceiverClientFactory.releaseClientForReadData(xceiverClient, false); xceiverClient = null; } @@ -382,6 +384,10 @@ public Duration getReadTimeout() { return readTimeout; } + private Object getName(StreamingReader reader) { + return reader != null ? reader : name; + } + /** * Implementation of a StreamObserver used to received and buffer streaming GRPC reads. */ @@ -461,9 +467,8 @@ ByteBuffer readFromQueue() throws IOException { final long blockOffset = readBlock.getOffset(); final long pos = getPos(); if (pos < blockOffset) { - // This should not happen, and if it does, we have a bug. - setFailedAndThrow(new IllegalStateException( - this + ": out of order, position " + pos + " < block offset " + blockOffset)); + // This can happen after seek, just drop the buffer + return null; } final long offset = pos - blockOffset; if (offset > 0) { From 8a37e8bfe25717b9accd168ddb67a6526b8fe1eb Mon Sep 17 00:00:00 2001 From: Tsz-Wo Nicholas Sze Date: Tue, 2 Jun 2026 11:47:27 -0700 Subject: [PATCH 2/7] Put back the proto to queue for seek. --- .../scm/storage/StreamBlockInputStream.java | 46 ++++++++++++++----- 1 file changed, 35 insertions(+), 11 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 e347e33e98d2..76f4d8087c5d 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 @@ -82,7 +82,7 @@ public class StreamBlockInputStream extends BlockExtendedInputStream { private XceiverClientFactory xceiverClientFactory; private XceiverClientGrpc xceiverClient; - private ByteBuffer buffer; + private ReadBuffer readBuffer; private long position = 0; private long requestedLength = 0; private StreamingReader streamingReader; @@ -136,7 +136,7 @@ public synchronized int read() throws IOException { if (!dataAvailableToRead(1, preRead)) { return EOF; } - int value = buffer.get(); + final int value = readBuffer.getByteBuffer().get(); advancePosition(1, preRead); return value; } @@ -159,6 +159,8 @@ synchronized int readFully(ByteBuffer targetBuf, boolean preRead) throws IOExcep if (!dataAvailableToRead(targetBuf.remaining(), preRead)) { break; } + + final ByteBuffer buffer = readBuffer.getByteBuffer(); int toCopy = Math.min(buffer.remaining(), targetBuf.remaining()); ByteBuffer tmpBuf = buffer.duplicate(); tmpBuf.limit(tmpBuf.position() + toCopy); @@ -179,7 +181,7 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th if (bufferHasRemaining()) { return true; } - buffer = streamingReader.read(length, preRead); + readBuffer = streamingReader.read(length, preRead); return bufferHasRemaining(); } @@ -192,6 +194,7 @@ private synchronized void advancePosition(long delta, boolean preRead) { } private synchronized boolean bufferHasRemaining() { + final ByteBuffer buffer = readBuffer.getByteBuffer(); return buffer != null && buffer.hasRemaining(); } @@ -213,7 +216,10 @@ public synchronized void seek(long pos) throws IOException { return; } LOG.debug("{}: seek {} -> {}", getName(streamingReader), position, pos); - buffer = null; + if (streamingReader != null) { + streamingReader.offerToQueue(readBuffer.getProto()); + } + readBuffer = null; position = pos; requestedLength = pos; } @@ -231,14 +237,13 @@ public synchronized void unbuffer() { } private synchronized void closeReader(String reason) { + readBuffer = null; if (streamingReader == null) { - buffer = null; return; } final StreamingReader reader = streamingReader; streamingReader = null; - buffer = null; LOG.debug("{} closeReader for {}", getName(reader), reason); reader.onCompleted(); @@ -388,6 +393,24 @@ private Object getName(StreamingReader reader) { return reader != null ? reader : name; } + static class ReadBuffer { + private final ReadBlockResponseProto proto; + private final ByteBuffer buffer; + + ReadBuffer(ReadBlockResponseProto proto, ByteBuffer buffer) { + this.proto = proto; + this.buffer = buffer; + } + + ReadBlockResponseProto getProto() { + return proto; + } + + ByteBuffer getByteBuffer() { + return buffer; + } + } + /** * Implementation of a StreamObserver used to received and buffer streaming GRPC reads. */ @@ -441,7 +464,7 @@ ReadBlockResponseProto poll() throws IOException { } } - private ByteBuffer read(int length, boolean preRead) throws IOException { + private ReadBuffer read(int length, boolean preRead) throws IOException { checkError(); if (future.isDone()) { return null; // Stream ended @@ -450,14 +473,15 @@ private ByteBuffer read(int length, boolean preRead) throws IOException { readBlock(length, preRead); while (true) { - final ByteBuffer buf = readFromQueue(); + final ReadBuffer readBuffer = readFromQueue(); + final ByteBuffer buf = readBuffer.getByteBuffer(); if (buf != null && buf.hasRemaining()) { - return buf; + return readBuffer; } } } - ByteBuffer readFromQueue() throws IOException { + ReadBuffer readFromQueue() throws IOException { final ReadBlockResponseProto readBlock = poll(); // 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. @@ -476,7 +500,7 @@ ByteBuffer readFromQueue() throws IOException { } LOG.debug("{}: return response positon {}, length {} (block offset {}, length {})", name, pos, dataBuffer.remaining(), blockOffset, data.size()); - return dataBuffer; + return new ReadBuffer(readBlock, dataBuffer); } private void releaseResources() { From 541bf946c59c8f4494ac063997e85e5477cc6845 Mon Sep 17 00:00:00 2001 From: Tsz-Wo Nicholas Sze Date: Tue, 2 Jun 2026 12:11:44 -0700 Subject: [PATCH 3/7] Fix checkstyle --- .../hadoop/hdds/scm/storage/StreamBlockInputStream.java | 6 +++--- 1 file changed, 3 insertions(+), 3 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 76f4d8087c5d..2367b43bf428 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 @@ -473,10 +473,10 @@ private ReadBuffer read(int length, boolean preRead) throws IOException { readBlock(length, preRead); while (true) { - final ReadBuffer readBuffer = readFromQueue(); - final ByteBuffer buf = readBuffer.getByteBuffer(); + final ReadBuffer read = readFromQueue(); + final ByteBuffer buf = read.getByteBuffer(); if (buf != null && buf.hasRemaining()) { - return readBuffer; + return read; } } } From 750c5ac5ceee7378dd58b168d844753a3a961a59 Mon Sep 17 00:00:00 2001 From: Tsz-Wo Nicholas Sze Date: Tue, 2 Jun 2026 14:08:35 -0700 Subject: [PATCH 4/7] Check readBuffer == null --- .../scm/storage/StreamBlockInputStream.java | 29 ++++++++----------- 1 file changed, 12 insertions(+), 17 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 2367b43bf428..7ac53dd411fb 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 @@ -178,11 +178,11 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th } initialize(); - if (bufferHasRemaining()) { + if (hasRemaining(readBuffer)) { return true; } readBuffer = streamingReader.read(length, preRead); - return bufferHasRemaining(); + return hasRemaining(readBuffer); } private synchronized void advancePosition(long delta, boolean preRead) { @@ -193,9 +193,8 @@ private synchronized void advancePosition(long delta, boolean preRead) { } } - private synchronized boolean bufferHasRemaining() { - final ByteBuffer buffer = readBuffer.getByteBuffer(); - return buffer != null && buffer.hasRemaining(); + private static boolean hasRemaining(ReadBuffer read) { + return read != null && read.getByteBuffer().hasRemaining(); } @Override @@ -216,7 +215,7 @@ public synchronized void seek(long pos) throws IOException { return; } LOG.debug("{}: seek {} -> {}", getName(streamingReader), position, pos); - if (streamingReader != null) { + if (streamingReader != null && readBuffer != null) { streamingReader.offerToQueue(readBuffer.getProto()); } readBuffer = null; @@ -456,9 +455,12 @@ ReadBlockResponseProto poll() throws IOException { } final long elapsedNanos = System.nanoTime() - startTime; - if (elapsedNanos >= readTimeoutNanos) { - setFailedAndThrow(new TimeoutIOException( - "Timed out waiting for response after " + readTimeout)); + if (elapsedNanos >= readTimeoutNanos && !future.isDone()) { + final TimeoutIOException e = new TimeoutIOException( + this + ": Failed to poll a response, timed out " + readTimeout); + if (setFailed(e)) { + throw e; + } return null; } } @@ -474,8 +476,7 @@ private ReadBuffer read(int length, boolean preRead) throws IOException { while (true) { final ReadBuffer read = readFromQueue(); - final ByteBuffer buf = read.getByteBuffer(); - if (buf != null && buf.hasRemaining()) { + if (hasRemaining(read)) { return read; } } @@ -556,12 +557,6 @@ StreamingReadResponse getResponse() { return response.get(); } - private void setFailedAndThrow(T throwable) throws T { - if (setFailed(throwable)) { - throw throwable; - } - } - private boolean setFailed(Throwable throwable) { final boolean completed = future.completeExceptionally(throwable); if (!completed) { From 70d76485c79a1307e64a7c27ead7ec842efeac99 Mon Sep 17 00:00:00 2001 From: Tsz-Wo Nicholas Sze Date: Fri, 5 Jun 2026 11:53:38 -0700 Subject: [PATCH 5/7] Reuse read buffer --- .../scm/storage/StreamBlockInputStream.java | 75 +++++++++++-------- 1 file changed, 44 insertions(+), 31 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 7ac53dd411fb..473f5289abc1 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 @@ -178,11 +178,11 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th } initialize(); - if (hasRemaining(readBuffer)) { - return true; + if (!hasRemaining(readBuffer)) { + readBuffer = streamingReader.read(length, preRead); } - readBuffer = streamingReader.read(length, preRead); - return hasRemaining(readBuffer); + Preconditions.assertTrue(hasRemaining(readBuffer)); + return true; } private synchronized void advancePosition(long delta, boolean preRead) { @@ -215,14 +215,39 @@ public synchronized void seek(long pos) throws IOException { return; } LOG.debug("{}: seek {} -> {}", getName(streamingReader), position, pos); - if (streamingReader != null && readBuffer != null) { - streamingReader.offerToQueue(readBuffer.getProto()); - } - readBuffer = null; + readBuffer = reuseReadBuffer(readBuffer, pos); position = pos; requestedLength = pos; } + static ReadBuffer reuseReadBuffer(ReadBuffer previous, long blockOffset) { + if (previous != null) { + final ByteBuffer buffer = getByteBuffer(previous.getProto(), blockOffset); + if (buffer != null && buffer.hasRemaining()) { + previous.getByteBuffer().position(buffer.position()); + Preconditions.assertSame(buffer.remaining(), previous.getByteBuffer().remaining(), "remaining"); + return previous; + } + } + return null; + } + + static ByteBuffer getByteBuffer(ReadBlockResponseProto proto, long blockOffset) { + final ByteBuffer buffer = proto.getData().asReadOnlyByteBuffer(); + // Adjust buffer position since the server always returns data starting at checksum boundary. + final long protoOffset = proto.getOffset(); + if (blockOffset < protoOffset) { + // This can happen after seek, just drop it for now + // TODO: consider to cache the proto, which will be useful when seeking back. + return null; + } + final long offset = blockOffset - protoOffset; + if (offset > 0) { + buffer.position(Math.toIntExact(Math.min(offset, buffer.limit()))); + } + return buffer; + } + @Override // The seekable interface indicates that seekToNewSource should seek to a new source of the data, // ie a different datanode. This is not supported for now. @@ -408,6 +433,13 @@ ReadBlockResponseProto getProto() { ByteBuffer getByteBuffer() { return buffer; } + + @Override + public String toString() { + return "ReadBuffer: offset=" + proto.getOffset() + + ", dataSize=" + proto.getData().size() + + ", buffer=" + buffer; + } } /** @@ -475,35 +507,16 @@ private ReadBuffer read(int length, boolean preRead) throws IOException { readBlock(length, preRead); while (true) { - final ReadBuffer read = readFromQueue(); + final ReadBlockResponseProto proto = poll(); + final ByteBuffer buffer = getByteBuffer(proto, getPos()); + final ReadBuffer read = buffer != null ? new ReadBuffer(proto, buffer) : null; if (hasRemaining(read)) { + LOG.debug("{}: read(length={}, preRead={}) returns {}", name, length, preRead, read); return read; } } } - ReadBuffer readFromQueue() throws IOException { - final ReadBlockResponseProto readBlock = poll(); - // 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(); - final long pos = getPos(); - if (pos < blockOffset) { - // This can happen after seek, just drop the buffer - return null; - } - final long offset = pos - blockOffset; - if (offset > 0) { - dataBuffer.position(Math.toIntExact(Math.min(offset, dataBuffer.limit()))); - } - LOG.debug("{}: return response positon {}, length {} (block offset {}, length {})", - name, pos, dataBuffer.remaining(), blockOffset, data.size()); - return new ReadBuffer(readBlock, dataBuffer); - } - private void releaseResources() { if (semaphoreReleased.compareAndSet(false, true)) { releaseStreamResources(); From ab6302ba08c13544286d452059fc3de9948edada Mon Sep 17 00:00:00 2001 From: Tsz-Wo Nicholas Sze Date: Mon, 8 Jun 2026 12:34:10 -0700 Subject: [PATCH 6/7] Address review comments. --- .../scm/storage/MultipartInputStream.java | 4 +- .../scm/storage/StreamBlockInputStream.java | 9 ++- .../rpc/read/TestStreamBlockInputStream.java | 56 +++++++++++++++---- .../ozone/client/rpc/read/TestStreamRead.java | 1 + 4 files changed, 56 insertions(+), 14 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java index 221a48be828d..47ddc99b8e94 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java @@ -174,7 +174,9 @@ public synchronized void seek(long pos) throws IOException { } // Reset the previous partStream's position - partStreams.get(prevPartIndex).seek(0); + if (prevPartIndex != partIndex) { + partStreams.get(prevPartIndex).seek(0); + } // Reset all the partStreams above the partIndex. We do this to reset // any previous reads which might have updated the higher part 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 473f5289abc1..f85de36b4436 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 @@ -186,7 +186,7 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th } private synchronized void advancePosition(long delta, boolean preRead) { - LOG.debug("{}: advance {} -> {}", getName(streamingReader), position, position + delta); + LOG.trace("{}: advance {} -> {}", getName(streamingReader), position, position + delta); position += delta; if (preRead && position >= blockLength) { closeReader("advancePosition"); @@ -508,6 +508,9 @@ private ReadBuffer read(int length, boolean preRead) throws IOException { while (true) { final ReadBlockResponseProto proto = poll(); + if (proto == null) { + return null; + } final ByteBuffer buffer = getByteBuffer(proto, getPos()); final ReadBuffer read = buffer != null ? new ReadBuffer(proto, buffer) : null; if (hasRemaining(read)) { @@ -597,9 +600,9 @@ private void setCompleted() { } private void offerToQueue(ReadBlockResponseProto item) { - if (LOG.isDebugEnabled()) { + if (LOG.isTraceEnabled()) { final ContainerProtos.ChecksumData checksumData = item.getChecksumData(); - LOG.debug("{}: enqueue response offset {}, length {}, numChecksums {}, bytesPerChecksum={}", + LOG.trace("{}: enqueue response offset {}, length {}, numChecksums {}, bytesPerChecksum={}", name, item.getOffset(), item.getData().size(), checksumData.getChecksumsList().size(), checksumData.getBytesPerChecksum()); } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java index 44b753210d91..ac3bf004fb3d 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Random; import java.util.concurrent.ThreadLocalRandom; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -61,7 +62,7 @@ public class TestStreamBlockInputStream extends TestInputStreamBase { GenericTestUtils.setLogLevel(LoggerFactory.getLogger("SCMHATransactionMonitor"), Level.ERROR); GenericTestUtils.setLogLevel(GrpcXceiverService.class, Level.ERROR); -// GenericTestUtils.setLogLevel(LoggerFactory.getLogger(StreamBlockInputStream.class), Level.TRACE); +// GenericTestUtils.setLogLevel(StreamBlockInputStream.class, Level.DEBUG); // GenericTestUtils.setLogLevel(LoggerFactory.getLogger(XceiverClientGrpc.class), Level.TRACE); } @@ -81,7 +82,7 @@ void testReadKey() throws Exception { OzoneConfiguration conf = cluster.getConf(); runTestReadKey(DATA_LENGTH, false, conf); - for (int i = 0; i < 3; i++) { + for (int i = 0; i < 2; i++) { final int keyLength = DATA_LENGTH + ThreadLocalRandom.current().nextInt(DATA_LENGTH); runTestReadKey(keyLength, true, conf); } @@ -185,13 +186,25 @@ void assertData(int pos, int length, ByteBuffer buffer) { } @Test - void testAll() throws Exception { + void testAllWithPreRead() throws Exception { + runTestAll(true); + } + + @Test + void testAllWithoutPreRead() throws Exception { + runTestAll(false); + } + + void runTestAll(boolean preRead) throws Exception { try (MiniOzoneCluster cluster = newCluster()) { cluster.waitForClusterToBeReady(); OzoneConfiguration conf = cluster.getConf(); OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); clientConfig.setStreamReadBlock(true); + if (!preRead) { + clientConfig.setStreamReadPreReadSize(0); + } OzoneConfiguration copy = new OzoneConfiguration(conf); copy.setFromObject(clientConfig); String keyName = getNewKeyName(); @@ -250,15 +263,38 @@ private void testReadKeyFully(String key) throws Exception { } } + void assertSeekRead(KeyInputStream in, int position) throws IOException { + in.seek(position); + int b = in.read(); + assertEquals(inputData[position], (byte) b, "Read data is not same as written data at index " + position); + } + + private void runTestSeek(KeyInputStream in, int seekSize, Random random) throws IOException { + LOG.info("runTestSeek: seekSize={}", seekSize); + for (int i = 0; i < 100; i++) { + int position = random.nextInt(seekSize); + assertSeekRead(in, position); + } + + for (int position = 0; position < DATA_LENGTH; position += random.nextInt(seekSize)) { + assertSeekRead(in, position); + } + + for (int position = DATA_LENGTH - 1; position >= 0; position -= random.nextInt(seekSize)) { + assertSeekRead(in, position); + } + assertSeekRead(in, 0); + } + private void testSeek(String key) throws IOException { - java.util.Random random = new java.util.Random(); + final Random random = new Random(); try (KeyInputStream keyInputStream = bucket.getKeyInputStream(key)) { - for (int i = 0; i < 100; i++) { - int position = random.nextInt(DATA_LENGTH); - keyInputStream.seek(position); - int b = keyInputStream.read(); - assertEquals(inputData[position], (byte) b, "Read data is not same as written data at index " + position); - } + runTestSeek(keyInputStream, CHUNK_SIZE / 8, random); + runTestSeek(keyInputStream, CHUNK_SIZE, random); + runTestSeek(keyInputStream, BLOCK_SIZE, random); + runTestSeek(keyInputStream, DATA_LENGTH, random); + + // error cases StreamBlockInputStream blockStream = (StreamBlockInputStream) keyInputStream.getPartStreams().get(0); long length = blockStream.getLength(); blockStream.seek(10); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java index 9fc217b6df3a..d3bacb3e84c6 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java @@ -74,6 +74,7 @@ public class TestStreamRead { GenericTestUtils.setLogLevel(LoggerFactory.getLogger("ExpiredContainerReplicaOpScrubber"), Level.ERROR); GenericTestUtils.setLogLevel(LoggerFactory.getLogger("SCMHATransactionMonitor"), Level.ERROR); GenericTestUtils.setLogLevel(LoggerFactory.getLogger(CodecBuffer.class), Level.ERROR); +// GenericTestUtils.setLogLevel(StreamBlockInputStream.class, Level.DEBUG); } static final int CHUNK_SIZE = 1 << 20; // 1MB From cef6b47d097f66382c9ca3df5d370b60d64f8c7b Mon Sep 17 00:00:00 2001 From: Tsz-Wo Nicholas Sze Date: Wed, 17 Jun 2026 09:58:23 -0700 Subject: [PATCH 7/7] Fix the bugs discovered by Cursor. --- .../hadoop/hdds/scm/storage/StreamBlockInputStream.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 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 f85de36b4436..7ef083cbdbd2 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 @@ -181,8 +181,7 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th if (!hasRemaining(readBuffer)) { readBuffer = streamingReader.read(length, preRead); } - Preconditions.assertTrue(hasRemaining(readBuffer)); - return true; + return hasRemaining(readBuffer); } private synchronized void advancePosition(long delta, boolean preRead) { @@ -217,7 +216,11 @@ public synchronized void seek(long pos) throws IOException { LOG.debug("{}: seek {} -> {}", getName(streamingReader), position, pos); readBuffer = reuseReadBuffer(readBuffer, pos); position = pos; - requestedLength = pos; + if (readBuffer == null) { + // Only rewind the request high-watermark when the buffered (already requested/served) data cannot be reused; + // otherwise we would re-request data that is still buffered. + requestedLength = pos; + } } static ReadBuffer reuseReadBuffer(ReadBuffer previous, long blockOffset) {