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..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 @@ -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,25 @@ public synchronized void initialize() throws IOException { } private void refreshBlockInfo(IOException cause) throws IOException { - refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); + lock.lock(); + try { + refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); + } finally { + lock.unlock(); + } + } + + private void refreshBlockInfoForPositionRead(IOException cause, Pipeline pipeline) throws IOException { + lock.lock(); + try { + if (failedPipeline != pipeline) { + refreshBlockInfo(cause, blockID, pipelineRef, tokenRef, refreshFunction); + failedPipeline = pipeline; + } + + } finally { + lock.unlock(); + } } /** @@ -289,7 +313,106 @@ 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(); + } + + 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; + 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; + } + + 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; + } + + // 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) { + if (totalReadLen == 0) { + throw new EOFException( + "EOF encountered at pos: " + pos + " for block: " + blockID); + } + 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)) { + handlePositionReadError(e, pipelineRef.get()); + continue; + } else { + throw e; + } + } catch (IOException ex) { + if (shouldRetryRead(ex, retryPolicy, ++innerRetries)) { + if (isConnectivityIssue(ex)) { + handlePositionReadError(ex, pipelineRef.get()); + } 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; + totalReadLen += numBytesRead; + chunkIdx++; + } + return true; } @Override @@ -464,9 +587,14 @@ 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 +646,35 @@ 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); + 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, pipeline); + } finally { + lock.unlock(); + } } 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..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 @@ -26,6 +26,7 @@ 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.ByteBufferReadable; @@ -101,11 +102,13 @@ public class ChunkInputStream extends InputStream private static final int EOF = -1; private final List validators; + private final ReentrantLock lock; + 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 +117,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 +313,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 +423,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 +440,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 +753,99 @@ public ByteBuffer[] getCachedBuffers() { public ChunkInfo getChunkInfo() { return chunkInfo; } + + 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(); + 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; + } + + 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; + } finally { + if (xceiverClientFactory != null && pair.getLeft() != null) { + xceiverClientFactory.releaseClientForReadData(pair.getLeft(), false); + } + } + } + + public void readFully(long l, ByteBuffer byteBuffer) throws IOException { + read(l, byteBuffer); + } + + 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/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-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..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 @@ -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,9 @@ 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..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 @@ -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,48 @@ public void connectsToNewPipeline() throws Exception { verify(newToken).encodeToUrlString(); } } + + @Test + public void testPositionedRead() throws Exception { + ByteBuffer byteBuffer = ByteBuffer.allocate(50); + int bytesRead = chunkStream.read(30, byteBuffer); + assertEquals(50, bytesRead); + byte[] expected = Arrays.copyOfRange(chunkData, 30, 80); + 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); + byte[] expected = Arrays.copyOfRange(chunkData, 50, 90); + 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()); + } } 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(); } }