From 9f551e841fe06846f93603631897b82db6c2e6ab Mon Sep 17 00:00:00 2001 From: chungen0126 Date: Tue, 14 Jul 2026 04:14:59 +0800 Subject: [PATCH 1/7] HDDS-15734. Implementing position read in BlockInputStream --- .../hdds/scm/storage/BlockInputStream.java | 130 ++++++++++++++++-- .../hdds/scm/storage/ChunkInputStream.java | 126 +++++++++++++++-- .../hdds/scm/storage/ExtendedInputStream.java | 1 + .../scm/storage/DummyChunkInputStream.java | 15 +- .../scm/storage/TestBlockInputStream.java | 18 +++ .../scm/storage/TestChunkInputStream.java | 26 +++- 6 files changed, 287 insertions(+), 29 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index 6f6b513422f7..617614038322 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java @@ -22,10 +22,12 @@ import java.io.EOFException; import java.io.IOException; import java.io.InputStream; +import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData; @@ -69,7 +71,7 @@ public class BlockInputStream extends BlockExtendedInputStream { private final boolean verifyChecksum; private XceiverClientFactory xceiverClientFactory; private XceiverClientSpi xceiverClient; - private boolean initialized = false; + private volatile boolean initialized = false; // TODO: do we need to change retrypolicy based on exception. private final RetryPolicy retryPolicy; @@ -108,6 +110,10 @@ public class BlockInputStream extends BlockExtendedInputStream { private BlockData blockData; + private Pipeline failedPipeline; + + private ReentrantLock lock = new ReentrantLock(); + public BlockInputStream( BlockLocationInfo blockInfo, Pipeline pipeline, @@ -209,7 +215,16 @@ public synchronized void initialize() throws IOException { } private void refreshBlockInfo(IOException cause) throws IOException { - refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); + lock.lock(); + try { + if (failedPipeline != pipelineRef.get()) { + refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); + failedPipeline = pipelineRef.get(); + } + + } finally { + lock.unlock(); + } } /** @@ -289,7 +304,85 @@ protected synchronized void addStream(ChunkInfo chunkInfo) { protected ChunkInputStream createChunkInputStream(ChunkInfo chunkInfo) { return new ChunkInputStream(chunkInfo, blockID, - xceiverClientFactory, pipelineRef::get, verifyChecksum, tokenRef::get); + xceiverClientFactory, pipelineRef::get, verifyChecksum, tokenRef::get, lock); + } + + @Override + public boolean readFully(long pos, ByteBuffer buffer) throws IOException { + Preconditions.checkArgument(buffer != null); + if (!initialized) { + initialize(); + } + checkOpen(); + int len = buffer.remaining(); + int innerRetries = 0; + int chunkIdx = Arrays.binarySearch(chunkOffsets, pos); + if (chunkIdx < 0) { + // Binary search returns -insertionPoint - 1 if element is not present + // in the array. insertionPoint is the point at which element would be + // inserted in the sorted array. We need to adjust the chunkIndex + // accordingly so that chunkIndex = insertionPoint - 1 + chunkIdx = -chunkIdx - 2; + } + + while (len > 0) { + if (chunkIdx >= chunkStreams.size()) { + return true; + } + + // Get the current chunkStream and read data from it + ChunkInputStream current = chunkStreams.get(chunkIdx); + long offsetInChunk = pos - chunkOffsets[chunkIdx]; + int numBytesToRead = Math.min(len, (int)(current.getLength() - offsetInChunk)); + if (numBytesToRead <= 0) { + return true; + } + int numBytesRead; + int bufferLimit = buffer.limit(); + try { + if (numBytesToRead < len) { + buffer.limit(buffer.position() + numBytesToRead); + } + numBytesRead = current.read(offsetInChunk, buffer); + innerRetries = 0; + + } catch (SCMSecurityException ex) { + throw ex; + } catch (StorageContainerException e) { + if (shouldRetryRead(e, retryPolicy, ++innerRetries)) { + handleReadError(e); + continue; + } else { + throw e; + } + } catch (IOException ex) { + if (shouldRetryRead(ex, retryPolicy, ++innerRetries)) { + if (isConnectivityIssue(ex)) { + handleReadError(ex); + } else { + current.releaseClient(); + } + continue; + } else { + throw ex; + } + } finally { + buffer.limit(bufferLimit); + } + if (numBytesRead != numBytesToRead) { + // This implies that there is either data loss or corruption in the + // chunk entries. Even EOF in the current stream would be covered in + // this case. + throw new IOException(String.format( + "Inconsistent read for chunkName=%s length=%d numBytesToRead= %d " + + "numBytesRead=%d", current.getChunkName(), current.getLength(), + numBytesToRead, numBytesRead)); + } + len -= numBytesRead; + pos += numBytesRead; + chunkIdx++; + } + return true; } @Override @@ -464,10 +557,15 @@ public synchronized void close() { } private void releaseClient() { - if (xceiverClientFactory != null && xceiverClient != null) { - xceiverClientFactory.releaseClientForReadData(xceiverClient, false); - xceiverClient = null; - } + lock.lock(); + try { + if (xceiverClientFactory != null && xceiverClient != null) { + xceiverClientFactory.releaseClientForReadData(xceiverClient, false); + xceiverClient = null; + } + } finally { + lock.unlock(); + } } /** @@ -518,15 +616,19 @@ private synchronized void storePosition() { } private void handleReadError(IOException cause) throws IOException { - releaseClient(); - final List inputStreams = this.chunkStreams; - if (inputStreams != null) { - for (ChunkInputStream is : inputStreams) { - is.releaseClient(); + lock.lock(); + try { + releaseClient(); + final List inputStreams = this.chunkStreams; + if (inputStreams != null) { + for (ChunkInputStream is : inputStreams) { + is.releaseClient(); + } } + refreshBlockInfo(cause); + } finally { + lock.unlock(); } - - refreshBlockInfo(cause); } public synchronized List getChunkStreams() { diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java index 22917ce4b6c7..5420b706c7d2 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java @@ -26,8 +26,10 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.List; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.Supplier; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.fs.ByteBufferPositionedReadable; import org.apache.hadoop.fs.ByteBufferReadable; import org.apache.hadoop.fs.CanUnbuffer; import org.apache.hadoop.fs.Seekable; @@ -55,7 +57,7 @@ * instances. */ public class ChunkInputStream extends InputStream - implements Seekable, CanUnbuffer, ByteBufferReadable { + implements Seekable, CanUnbuffer, ByteBufferReadable, ByteBufferPositionedReadable { private final ChunkInfo chunkInfo; private final long length; @@ -101,11 +103,13 @@ public class ChunkInputStream extends InputStream private static final int EOF = -1; private final List validators; + private ReentrantLock lock = new ReentrantLock(); + ChunkInputStream(ChunkInfo chunkInfo, BlockID blockId, XceiverClientFactory xceiverClientFactory, Supplier pipelineSupplier, boolean verifyChecksum, - Supplier> tokenSupplier) { + Supplier> tokenSupplier, ReentrantLock lock) { this.chunkInfo = chunkInfo; this.length = chunkInfo.getLen(); this.blockID = blockId; @@ -114,6 +118,7 @@ public class ChunkInputStream extends InputStream this.verifyChecksum = verifyChecksum; this.tokenSupplier = tokenSupplier; validators = ContainerProtocolCalls.toValidatorList(this::validateChunk); + this.lock = lock != null ? lock : new ReentrantLock(); } public synchronized long getRemaining() { @@ -309,9 +314,14 @@ private void updateDatanodeBlockId(Pipeline pipeline) throws IOException { */ protected synchronized void acquireClient() throws IOException { if (xceiverClientFactory != null && xceiverClient == null) { - Pipeline pipeline = pipelineSupplier.get(); - xceiverClient = xceiverClientFactory.acquireClientForReadData(pipeline); - updateDatanodeBlockId(pipeline); + lock.lock(); + try { + Pipeline pipeline = pipelineSupplier.get(); + xceiverClient = xceiverClientFactory.acquireClientForReadData(pipeline); + updateDatanodeBlockId(pipeline); + } finally { + lock.unlock(); + } } } @@ -414,9 +424,7 @@ private synchronized void readChunkFromContainer(int len) throws IOException { private void readChunkDataIntoBuffers(ChunkInfo readChunkInfo) throws IOException { - buffers = readChunk(readChunkInfo); - buffersSize = readChunkInfo.getLen(); - + buffers = readChunk(xceiverClient, readChunkInfo, datanodeBlockID); buffersSize = readChunkInfo.getLen(); bufferOffsets = new long[buffers.length]; int tempOffset = 0; for (int i = 0; i < buffers.length; i++) { @@ -433,11 +441,11 @@ private void readChunkDataIntoBuffers(ChunkInfo readChunkInfo) * Send RPC call to get the chunk from the container. */ @VisibleForTesting - protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) + protected ByteBuffer[] readChunk( + XceiverClientSpi client, ChunkInfo readChunkInfo, ContainerProtos.DatanodeBlockID dnBlockID) throws IOException { - ReadChunkResponseProto readChunkResponse = - ContainerProtocolCalls.readChunk(xceiverClient, readChunkInfo, datanodeBlockID, validators, + ContainerProtocolCalls.readChunk(client, readChunkInfo, dnBlockID, validators, tokenSupplier.get()); if (readChunkResponse.hasData()) { @@ -746,4 +754,100 @@ public ByteBuffer[] getCachedBuffers() { public ChunkInfo getChunkInfo() { return chunkInfo; } + + @Override + public int read(long pos, ByteBuffer buffer) throws IOException { + Preconditions.checkArgument(buffer != null); + int len = buffer.remaining(); + if (len == 0) { + return 0; + } + + Pair pair = getClientAndUpdateBlock(); + + int total = 0; + long adjustedBuffersOffset, adjustedBuffersLen; + if (verifyChecksum) { + // Adjust the chunk offset and length to include required checksum + // boundaries + Pair adjustedOffsetAndLength = + computeChecksumBoundaries(pos, len); + adjustedBuffersOffset = adjustedOffsetAndLength.getLeft(); + adjustedBuffersLen = adjustedOffsetAndLength.getRight(); + } else { + // Read from the startByteIndex + adjustedBuffersOffset = pos; + adjustedBuffersLen = len; + } + + final ChunkInfo readChunkInfo = ChunkInfo.newBuilder(chunkInfo) + .setOffset(chunkInfo.getOffset() + adjustedBuffersOffset) + .setLen(adjustedBuffersLen) + .build(); + + ByteBuffer[] readBuffers = readChunk(pair.getLeft(), readChunkInfo, pair.getRight()); + + if (readBuffers == null) { + return EOF; + } + int bufferIdx = 0; + long skipLen = pos - adjustedBuffersOffset; + while (skipLen > 0 && bufferIdx < readBuffers.length) { + ByteBuffer readBuf = readBuffers[bufferIdx]; + if (readBuf.remaining() <= skipLen) { + skipLen -= readBuf.remaining(); + bufferIdx++; + } else { + readBuf.position(readBuf.position() + (int) skipLen); + skipLen = 0; + } + } + while (len > 0) { + if (bufferIdx >= readBuffers.length) { + break; + } + ByteBuffer readBuf = readBuffers[bufferIdx]; + int available = Math.min(len, readBuf.remaining()); + + ByteBuffer tmpBuf = readBuf.duplicate(); + tmpBuf.limit(tmpBuf.position() + available); + buffer.put(tmpBuf); + readBuf.position(tmpBuf.position()); + + len -= available; + total += available; + bufferIdx++; + } + return total; + } + + @Override + public void readFully(long l, ByteBuffer byteBuffer) throws IOException { + int bytesRead = read(l, byteBuffer); + if (bytesRead < byteBuffer.capacity()) { + throw new EOFException("EOF encountered at pos: " + l + " for chunk: " + + chunkInfo.getChunkName()); + } + } + + protected Pair getClientAndUpdateBlock() + throws IOException { + ContainerProtos.DatanodeBlockID.Builder builder = blockID.getDatanodeBlockIDProtobufBuilder(); + XceiverClientSpi client = null; + if (xceiverClientFactory != null) { + lock.lock(); + try { + Pipeline pipeline = pipelineSupplier.get(); + client = xceiverClientFactory.acquireClientForReadData(pipeline); + DatanodeDetails closestNode = pipeline.getClosestNode(); + int replicaIdx = pipeline.getReplicaIndex(closestNode); + if (replicaIdx > 0) { + builder.setReplicaIndex(replicaIdx); + } + } finally { + lock.unlock(); + } + } + return Pair.of(client, builder.build()); + } } diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java index 75e483ad55ec..5064676c6415 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ExtendedInputStream.java @@ -102,6 +102,7 @@ public boolean hasCapability(String capability) { switch (StringUtils.toLowerCase(capability)) { case StreamCapabilities.READBYTEBUFFER: case StreamCapabilities.UNBUFFER: + case StreamCapabilities.VECTOREDIO: return true; default: return false; diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java index 1ef6327c2a14..19e88c4c922d 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java @@ -17,12 +17,16 @@ package org.apache.hadoop.hdds.scm.storage; +import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; +import org.apache.commons.lang3.tuple.Pair; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID; import org.apache.hadoop.hdds.scm.XceiverClientFactory; +import org.apache.hadoop.hdds.scm.XceiverClientSpi; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.ozone.common.utils.BufferUtils; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; @@ -43,12 +47,13 @@ public DummyChunkInputStream(ChunkInfo chunkInfo, boolean verifyChecksum, byte[] data, Pipeline pipeline) { super(chunkInfo, blockId, xceiverClientFactory, () -> pipeline, - verifyChecksum, () -> null); + verifyChecksum, () -> null, null); this.chunkData = data.clone(); } @Override - protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo) { + protected ByteBuffer[] readChunk( + XceiverClientSpi client, ChunkInfo readChunkInfo, DatanodeBlockID datanodeBlockID) { int offset = (int) readChunkInfo.getOffset(); int remainingToRead = (int) readChunkInfo.getLen(); @@ -87,4 +92,10 @@ protected void releaseClient() { public List getReadByteBuffers() { return readByteBuffers; } + + + @Override + protected Pair getClientAndUpdateBlock() throws IOException { + return Pair.of(null, null); + } } diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java index 1b5e7667e845..8ea070780463 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java @@ -469,4 +469,22 @@ private static Stream exceptionsTriggersRefresh() { new StatusException(Status.UNAVAILABLE)))) ); } + + @Test + public void testPositionedReadFully() throws Exception { + // 1. Read first full chunk (100 bytes) starting at offset 0 + ByteBuffer buffer1 = ByteBuffer.allocate(100); + assertTrue(blockStream.readFully(0, buffer1)); + matchWithInputData(buffer1.array(), 0, 100); + + // 2. Read crossing chunk boundary: start at offset 50 (middle of chunk 0), read 120 bytes (into chunk 1) + ByteBuffer buffer2 = ByteBuffer.allocate(120); + assertTrue(blockStream.readFully(50, buffer2)); + matchWithInputData(buffer2.array(), 50, 120); + + // 3. Read up to the end of the block: start at offset 400 (start of last chunk), read 50 bytes + ByteBuffer buffer3 = ByteBuffer.allocate(50); + assertTrue(blockStream.readFully(400, buffer3)); + matchWithInputData(buffer3.array(), 400, 50); + } } diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java index 248ea8655223..7c01e7a7eeca 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java @@ -30,6 +30,7 @@ import java.io.EOFException; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.List; import java.util.Random; import java.util.concurrent.atomic.AtomicLong; @@ -269,8 +270,7 @@ public void connectsToNewPipeline() throws Exception { ByteStringConversion::safeWrap)); try (ChunkInputStream subject = new ChunkInputStream(chunkInfo, blockID, - clientFactory, pipelineRef::get, false, tokenRef::get)) { - // WHEN + clientFactory, pipelineRef::get, false, tokenRef::get, null)) { // WHEN subject.unbuffer(); pipelineRef.set(newPipeline); tokenRef.set(newToken); @@ -284,4 +284,26 @@ public void connectsToNewPipeline() throws Exception { verify(newToken).encodeToUrlString(); } } + + @Test + public void testPositionedRead() throws Exception { + byte[] buffer = new byte[50]; + ByteBuffer byteBuffer = ByteBuffer.wrap(buffer); + int bytesRead = chunkStream.read(30, byteBuffer); + + assertEquals(50, bytesRead); + byte[] expected = Arrays.copyOfRange(chunkData, 30, 80); + assertArrayEquals(expected, buffer); + } + + @Test + public void testPositionedReadFully() throws Exception { + ByteBuffer byteBuffer = ByteBuffer.allocate(40); + chunkStream.readFully(50, byteBuffer); + byteBuffer.flip(); + byte[] actual = new byte[40]; + byteBuffer.get(actual); + byte[] expected = Arrays.copyOfRange(chunkData, 50, 90); + assertArrayEquals(expected, actual); + } } From a882c2335baf36f5d1faaff772952322d78e1830 Mon Sep 17 00:00:00 2001 From: chungen0126 Date: Tue, 14 Jul 2026 06:15:40 +0800 Subject: [PATCH 2/7] fix checkstyle --- .../org/apache/hadoop/hdds/scm/storage/BlockInputStream.java | 2 +- .../apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java | 1 - 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index 617614038322..ba620253d451 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java @@ -565,7 +565,7 @@ private void releaseClient() { } } finally { lock.unlock(); - } + } } /** diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java index 19e88c4c922d..33978ff6ce6b 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java @@ -93,7 +93,6 @@ public List getReadByteBuffers() { return readByteBuffers; } - @Override protected Pair getClientAndUpdateBlock() throws IOException { return Pair.of(null, null); From a17f876e7ec7ec2e24b93b0b7e2fa0c2873b3e96 Mon Sep 17 00:00:00 2001 From: chungen0126 Date: Wed, 15 Jul 2026 19:35:31 +0800 Subject: [PATCH 3/7] address comments --- .../hdds/scm/storage/BlockInputStream.java | 49 +++++++- .../hdds/scm/storage/ChunkInputStream.java | 112 +++++++++--------- 2 files changed, 98 insertions(+), 63 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index ba620253d451..f522bcebf25c 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java @@ -110,7 +110,7 @@ public class BlockInputStream extends BlockExtendedInputStream { private BlockData blockData; - private Pipeline failedPipeline; + private Pipeline currentPipeline; private ReentrantLock lock = new ReentrantLock(); @@ -217,9 +217,20 @@ public synchronized void initialize() throws IOException { private void refreshBlockInfo(IOException cause) throws IOException { lock.lock(); try { - if (failedPipeline != pipelineRef.get()) { + refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); + } finally { + lock.unlock(); + } + } + + private void refreshBlockInfoForPositionRead(IOException cause, Pipeline pipeline) throws IOException { + lock.lock(); + try { + if (currentPipeline == pipeline) { refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); - failedPipeline = pipelineRef.get(); + if (pipelineRef.get() != currentPipeline) { + currentPipeline = pipelineRef.get(); + } } } finally { @@ -325,8 +336,13 @@ public boolean readFully(long pos, ByteBuffer buffer) throws IOException { chunkIdx = -chunkIdx - 2; } + int totalReadLen = 0; while (len > 0) { if (chunkIdx >= chunkStreams.size()) { + if (totalReadLen == 0) { + throw new EOFException( + "EOF encountered at pos: " + pos + " for block: " + blockID); + } return true; } @@ -335,6 +351,10 @@ public boolean readFully(long pos, ByteBuffer buffer) throws IOException { long offsetInChunk = pos - chunkOffsets[chunkIdx]; int numBytesToRead = Math.min(len, (int)(current.getLength() - offsetInChunk)); if (numBytesToRead <= 0) { + if (totalReadLen == 0) { + throw new EOFException( + "EOF encountered at pos: " + pos + " for block: " + blockID); + } return true; } int numBytesRead; @@ -346,11 +366,11 @@ public boolean readFully(long pos, ByteBuffer buffer) throws IOException { numBytesRead = current.read(offsetInChunk, buffer); innerRetries = 0; - } catch (SCMSecurityException ex) { + } catch (SCMSecurityException ex) { throw ex; } catch (StorageContainerException e) { if (shouldRetryRead(e, retryPolicy, ++innerRetries)) { - handleReadError(e); + handlePositionReadError(e, pipelineRef.get()); continue; } else { throw e; @@ -358,7 +378,7 @@ public boolean readFully(long pos, ByteBuffer buffer) throws IOException { } catch (IOException ex) { if (shouldRetryRead(ex, retryPolicy, ++innerRetries)) { if (isConnectivityIssue(ex)) { - handleReadError(ex); + handlePositionReadError(ex, pipelineRef.get()); } else { current.releaseClient(); } @@ -380,6 +400,7 @@ public boolean readFully(long pos, ByteBuffer buffer) throws IOException { } len -= numBytesRead; pos += numBytesRead; + totalReadLen += numBytesRead; chunkIdx++; } return true; @@ -631,6 +652,22 @@ private void handleReadError(IOException cause) throws IOException { } } + private void handlePositionReadError(IOException cause, Pipeline pipeline) throws IOException { + lock.lock(); + try { + releaseClient(); + final List inputStreams = this.chunkStreams; + if (inputStreams != null) { + for (ChunkInputStream is : inputStreams) { + is.releaseClient(); + } + } + refreshBlockInfoForPositionRead(cause); + } finally { + lock.unlock(); + } + } + public synchronized List getChunkStreams() { return chunkStreams; } diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java index 5420b706c7d2..784fed3a2a12 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java @@ -29,7 +29,6 @@ import java.util.concurrent.locks.ReentrantLock; import java.util.function.Supplier; import org.apache.commons.lang3.tuple.Pair; -import org.apache.hadoop.fs.ByteBufferPositionedReadable; import org.apache.hadoop.fs.ByteBufferReadable; import org.apache.hadoop.fs.CanUnbuffer; import org.apache.hadoop.fs.Seekable; @@ -57,7 +56,7 @@ * instances. */ public class ChunkInputStream extends InputStream - implements Seekable, CanUnbuffer, ByteBufferReadable, ByteBufferPositionedReadable { + implements Seekable, CanUnbuffer, ByteBufferReadable { private final ChunkInfo chunkInfo; private final long length; @@ -103,7 +102,7 @@ public class ChunkInputStream extends InputStream private static final int EOF = -1; private final List validators; - private ReentrantLock lock = new ReentrantLock(); + private final ReentrantLock lock; ChunkInputStream(ChunkInfo chunkInfo, BlockID blockId, XceiverClientFactory xceiverClientFactory, @@ -755,7 +754,6 @@ public ChunkInfo getChunkInfo() { return chunkInfo; } - @Override public int read(long pos, ByteBuffer buffer) throws IOException { Preconditions.checkArgument(buffer != null); int len = buffer.remaining(); @@ -764,70 +762,70 @@ public int read(long pos, ByteBuffer buffer) throws IOException { } Pair pair = getClientAndUpdateBlock(); + try { + int total = 0; + long adjustedBuffersOffset, adjustedBuffersLen; + if (verifyChecksum) { + // Adjust the chunk offset and length to include required checksum + // boundaries + Pair adjustedOffsetAndLength = + computeChecksumBoundaries(pos, len); + adjustedBuffersOffset = adjustedOffsetAndLength.getLeft(); + adjustedBuffersLen = adjustedOffsetAndLength.getRight(); + } else { + // Read from the startByteIndex + adjustedBuffersOffset = pos; + adjustedBuffersLen = len; + } - int total = 0; - long adjustedBuffersOffset, adjustedBuffersLen; - if (verifyChecksum) { - // Adjust the chunk offset and length to include required checksum - // boundaries - Pair adjustedOffsetAndLength = - computeChecksumBoundaries(pos, len); - adjustedBuffersOffset = adjustedOffsetAndLength.getLeft(); - adjustedBuffersLen = adjustedOffsetAndLength.getRight(); - } else { - // Read from the startByteIndex - adjustedBuffersOffset = pos; - adjustedBuffersLen = len; - } - - final ChunkInfo readChunkInfo = ChunkInfo.newBuilder(chunkInfo) - .setOffset(chunkInfo.getOffset() + adjustedBuffersOffset) - .setLen(adjustedBuffersLen) - .build(); + final ChunkInfo readChunkInfo = ChunkInfo.newBuilder(chunkInfo) + .setOffset(chunkInfo.getOffset() + adjustedBuffersOffset) + .setLen(adjustedBuffersLen) + .build(); - ByteBuffer[] readBuffers = readChunk(pair.getLeft(), readChunkInfo, pair.getRight()); + ByteBuffer[] readBuffers = readChunk(pair.getLeft(), readChunkInfo, pair.getRight()); - if (readBuffers == null) { - return EOF; - } - int bufferIdx = 0; - long skipLen = pos - adjustedBuffersOffset; - while (skipLen > 0 && bufferIdx < readBuffers.length) { - ByteBuffer readBuf = readBuffers[bufferIdx]; - if (readBuf.remaining() <= skipLen) { - skipLen -= readBuf.remaining(); - bufferIdx++; - } else { - readBuf.position(readBuf.position() + (int) skipLen); - skipLen = 0; + if (readBuffers == null) { + return EOF; } - } - while (len > 0) { - if (bufferIdx >= readBuffers.length) { - break; + int bufferIdx = 0; + long skipLen = pos - adjustedBuffersOffset; + while (skipLen > 0 && bufferIdx < readBuffers.length) { + ByteBuffer readBuf = readBuffers[bufferIdx]; + if (readBuf.remaining() <= skipLen) { + skipLen -= readBuf.remaining(); + bufferIdx++; + } else { + readBuf.position(readBuf.position() + (int) skipLen); + skipLen = 0; + } } - ByteBuffer readBuf = readBuffers[bufferIdx]; - int available = Math.min(len, readBuf.remaining()); + while (len > 0) { + if (bufferIdx >= readBuffers.length) { + break; + } + ByteBuffer readBuf = readBuffers[bufferIdx]; + int available = Math.min(len, readBuf.remaining()); - ByteBuffer tmpBuf = readBuf.duplicate(); - tmpBuf.limit(tmpBuf.position() + available); - buffer.put(tmpBuf); - readBuf.position(tmpBuf.position()); + ByteBuffer tmpBuf = readBuf.duplicate(); + tmpBuf.limit(tmpBuf.position() + available); + buffer.put(tmpBuf); + readBuf.position(tmpBuf.position()); - len -= available; - total += available; - bufferIdx++; + len -= available; + total += available; + bufferIdx++; + } + return total; + } finally { + if (xceiverClientFactory != null && pair.getLeft() != null) { + xceiverClientFactory.releaseClientForReadData(pair.getLeft(), false); + } } - return total; } - @Override public void readFully(long l, ByteBuffer byteBuffer) throws IOException { - int bytesRead = read(l, byteBuffer); - if (bytesRead < byteBuffer.capacity()) { - throw new EOFException("EOF encountered at pos: " + l + " for chunk: " - + chunkInfo.getChunkName()); - } + read(l, byteBuffer); } protected Pair getClientAndUpdateBlock() From 4ef5a9286c57912d9d20d60c08c02dd1c0aa8d76 Mon Sep 17 00:00:00 2001 From: chungen0126 Date: Thu, 16 Jul 2026 00:53:53 +0800 Subject: [PATCH 4/7] address comments --- .../hadoop/hdds/scm/storage/BlockInputStream.java | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index f522bcebf25c..a00183d3ff1b 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java @@ -110,7 +110,7 @@ public class BlockInputStream extends BlockExtendedInputStream { private BlockData blockData; - private Pipeline currentPipeline; + private Pipeline failedPipeline; private ReentrantLock lock = new ReentrantLock(); @@ -226,11 +226,9 @@ private void refreshBlockInfo(IOException cause) throws IOException { private void refreshBlockInfoForPositionRead(IOException cause, Pipeline pipeline) throws IOException { lock.lock(); try { - if (currentPipeline == pipeline) { + if (failedPipeline != pipeline) { refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); - if (pipelineRef.get() != currentPipeline) { - currentPipeline = pipelineRef.get(); - } + failedPipeline = pipeline; } } finally { @@ -662,7 +660,7 @@ private void handlePositionReadError(IOException cause, Pipeline pipeline) throw is.releaseClient(); } } - refreshBlockInfoForPositionRead(cause); + refreshBlockInfoForPositionRead(cause, pipeline); } finally { lock.unlock(); } From c5ba62dc5c0c120a575bc62828a350966b186803 Mon Sep 17 00:00:00 2001 From: chungen0126 Date: Thu, 16 Jul 2026 19:57:14 +0800 Subject: [PATCH 5/7] make OzoneFSInputStream adopt BlockInputStream position read --- .../hdds/scm/storage/BlockInputStream.java | 12 +++++ .../scm/storage/MultipartInputStream.java | 49 ++++++++++++++++++- .../hadoop/fs/ozone/OzoneFSInputStream.java | 9 +++- 3 files changed, 68 insertions(+), 2 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index a00183d3ff1b..ac3f6023f593 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java @@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; +import org.apache.hadoop.fs.ByteBufferPositionedReadable; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; @@ -322,6 +323,17 @@ public boolean readFully(long pos, ByteBuffer buffer) throws IOException { if (!initialized) { initialize(); } + + if (pos < 0 || pos > length) { + if (pos == 0) { + // It is possible for length and pos to be zero in which case + // seek should return instead of throwing exception + return true; + } + throw new EOFException( + "EOF encountered at pos: " + pos + " for block: " + blockID); + } + checkOpen(); int len = buffer.remaining(); int innerRetries = 0; 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..04e06d035d83 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 @@ -190,7 +190,11 @@ public synchronized void seek(long pos) throws IOException { @Override public boolean readFully(long position, ByteBuffer buffer) throws IOException { if (!isStreamBlockInputStream) { - return false; + if (buffer.remaining() == 0) { + return true; + } + pRead(position, buffer); + return true; } final long oldPos = getPos(); @@ -209,6 +213,49 @@ int readImpl(InputStream inputStream) throws IOException { return true; } + public void pRead(long offset, ByteBuffer buffer) throws IOException { + int partIdx = Arrays.binarySearch(partOffsets, offset); + if (partIdx < 0) { + partIdx = -partIdx - 2; + } + + int len = buffer.remaining(); + while (len > 0) { + if (partIdx < 0 || partIdx >= partStreams.size()) { + throw new EOFException("EOF encountered at pos: " + offset + " for key: " + key); + } + + PartInputStream current = partStreams.get(partIdx); + if (!(current instanceof ExtendedInputStream)) { + throw new IOException("Positioned read is not supported by stream type: " + + current.getClass().getName()); + } + ExtendedInputStream extendedStream = (ExtendedInputStream) current; + + long offsetInPart = offset - partOffsets[partIdx]; + int numBytesToRead = Math.min(len, (int) (current.getLength() - offsetInPart)); + if (numBytesToRead <= 0) { + throw new EOFException("EOF encountered at pos: " + offset + " for key: " + key); + } + + int bufferLimit = buffer.limit(); + try { + if (numBytesToRead < len) { + buffer.limit(buffer.position() + numBytesToRead); + } + if (!extendedStream.readFully(offsetInPart, buffer)) { + throw new IOException("Positioned read failed on part stream"); + } + } finally { + buffer.limit(bufferLimit); + } + + len -= numBytesToRead; + offset += numBytesToRead; + partIdx++; + } + } + public synchronized void initialize() throws IOException { // Pre-check that the stream has not been intialized already if (initialized) { diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java index e640c1e6d175..b16dc0eb512f 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/OzoneFSInputStream.java @@ -171,7 +171,14 @@ public int read(long position, ByteBuffer buf) throws IOException { } if (inputStream instanceof ExtendedInputStream) { final int remainingBeforeRead = buf.remaining(); - if (((ExtendedInputStream) inputStream).readFully(position, buf)) { + try { + if (((ExtendedInputStream) inputStream).readFully(position, buf)) { + return remainingBeforeRead - buf.remaining(); + } + } catch (EOFException e) { + if (remainingBeforeRead - buf.remaining() == 0) { + return -1; + } return remainingBeforeRead - buf.remaining(); } } From ec3ee4c1054f67c27b30f4bbfa32d164074bbe32 Mon Sep 17 00:00:00 2001 From: chungen0126 Date: Thu, 16 Jul 2026 21:14:09 +0800 Subject: [PATCH 6/7] fix checkstyle --- .../org/apache/hadoop/hdds/scm/storage/BlockInputStream.java | 1 - 1 file changed, 1 deletion(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index ac3f6023f593..6190d40fd2a9 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java @@ -29,7 +29,6 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Function; -import org.apache.hadoop.fs.ByteBufferPositionedReadable; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; From 0c294de7864a048c533d88edadfb4e56e681c4de Mon Sep 17 00:00:00 2001 From: chungen0126 Date: Fri, 17 Jul 2026 10:45:40 +0800 Subject: [PATCH 7/7] update tests --- .../scm/storage/TestChunkInputStream.java | 38 +++++++++++++++---- 1 file changed, 30 insertions(+), 8 deletions(-) diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java index 7c01e7a7eeca..4372632b220a 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java @@ -287,23 +287,45 @@ public void connectsToNewPipeline() throws Exception { @Test public void testPositionedRead() throws Exception { - byte[] buffer = new byte[50]; - ByteBuffer byteBuffer = ByteBuffer.wrap(buffer); + ByteBuffer byteBuffer = ByteBuffer.allocate(50); int bytesRead = chunkStream.read(30, byteBuffer); - assertEquals(50, bytesRead); byte[] expected = Arrays.copyOfRange(chunkData, 30, 80); - assertArrayEquals(expected, buffer); + assertArrayEquals(expected, byteBuffer.array()); + + // Read backward + byteBuffer = ByteBuffer.allocate(50); + bytesRead = chunkStream.read(10, byteBuffer); + assertEquals(50, bytesRead); + expected = Arrays.copyOfRange(chunkData, 10, 60); + assertArrayEquals(expected, byteBuffer.array()); + + // Read forward + byteBuffer = ByteBuffer.allocate(50); + bytesRead = chunkStream.read(90, byteBuffer); + assertEquals(10, bytesRead); + expected = new byte[50]; + System.arraycopy(chunkData, 90, expected, 0, 10); + assertArrayEquals(expected, byteBuffer.array()); } @Test public void testPositionedReadFully() throws Exception { ByteBuffer byteBuffer = ByteBuffer.allocate(40); chunkStream.readFully(50, byteBuffer); - byteBuffer.flip(); - byte[] actual = new byte[40]; - byteBuffer.get(actual); byte[] expected = Arrays.copyOfRange(chunkData, 50, 90); - assertArrayEquals(expected, actual); + assertArrayEquals(expected, byteBuffer.array()); + + byteBuffer = ByteBuffer.allocate(50); + chunkStream.readFully(10, byteBuffer); + expected = Arrays.copyOfRange(chunkData, 10, 60); + assertArrayEquals(expected, byteBuffer.array()); + + + byteBuffer = ByteBuffer.allocate(50); + chunkStream.read(90, byteBuffer); + expected = new byte[50]; + System.arraycopy(chunkData, 90, expected, 0, 10); + assertArrayEquals(expected, byteBuffer.array()); } }