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 a9c2c8b2f0f..82f68643118 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
@@ -37,7 +37,9 @@
* The input stream for Ozone file system.
*
* TODO: Make inputStream generic for both rest and rpc clients
- * This class is not thread safe.
+ * Sequential reads are not thread safe. Positioned reads delegate to the
+ * underlying {@link ExtendedInputStream} when it supports them; otherwise they
+ * fall back to a synchronized seek-read-restore sequence.
*/
@InterfaceAudience.Private
@InterfaceStability.Evolving
@@ -46,6 +48,7 @@ public class OzoneFSInputStream extends FSInputStream
private final InputStream inputStream;
private final Statistics statistics;
+ private final Object positionedReadLock = new Object();
public OzoneFSInputStream(InputStream inputStream, Statistics statistics) {
this.inputStream = inputStream;
@@ -169,6 +172,9 @@ public int read(long position, ByteBuffer buf) throws IOException {
if (!buf.hasRemaining()) {
return 0;
}
+ if (position < 0) {
+ throw new EOFException("position is negative: " + position);
+ }
if (inputStream instanceof ExtendedInputStream) {
final int remainingBeforeRead = buf.remaining();
try {
@@ -180,6 +186,14 @@ public int read(long position, ByteBuffer buf) throws IOException {
}
}
+ // Fallback: stateful seek-read-restore on the shared cursor.
+ synchronized (positionedReadLock) {
+ return readAtPositionSeekRestore(position, buf);
+ }
+ }
+
+ private int readAtPositionSeekRestore(long position, ByteBuffer buf)
+ throws IOException {
long oldPos = this.getPos();
int bytesRead;
try {
@@ -211,4 +225,101 @@ public void readFully(long position, ByteBuffer buf) throws IOException {
}
}
}
+
+ /**
+ * Byte-array positioned read. Tries the native stateless {@link ExtendedInputStream#readFully}
+ * path first (via a zero-copy {@link ByteBuffer#wrap}). Falls back to a synchronized
+ * seek-read-restore using byte-array {@link #read(byte[], int, int)} so that the fallback
+ * works for any {@link Seekable} stream, not just those that also implement
+ * {@link org.apache.hadoop.fs.ByteBufferReadable}.
+ *
+ * {@link FSInputStream} synchronizes its inherited implementation on {@code this}, a different
+ * monitor from {@code positionedReadLock}; without this override the two APIs can interleave.
+ */
+ @Override
+ public int read(long position, byte[] buffer, int offset, int length) throws IOException {
+ // Validate before touching ByteBuffer so that null throws IAE (not NPE) and
+ // negative position propagates as EOFException rather than being swallowed.
+ validatePositionedReadArgs(position, buffer, offset, length);
+ if (length == 0) {
+ return 0;
+ }
+ if (inputStream instanceof ExtendedInputStream) {
+ final ByteBuffer buf = ByteBuffer.wrap(buffer, offset, length);
+ try {
+ if (((ExtendedInputStream) inputStream).readFully(position, buf)) {
+ final int bytesRead = length - buf.remaining();
+ // readFullyStateless can return true with bytesRead==0 for pos==length.
+ // Convert that to -1 to comply with PositionedReadable contract.
+ if (bytesRead == 0) {
+ return -1;
+ }
+ if (statistics != null) {
+ statistics.incrementBytesRead(bytesRead);
+ }
+ return bytesRead;
+ }
+ } catch (EOFException e) {
+ // pos < 0 was already rejected by validatePositionedReadArgs above,
+ // so this EOFException means pos >= stream length → return -1.
+ return -1;
+ }
+ }
+ synchronized (positionedReadLock) {
+ return readAtPositionSeekRestoreByteArray(position, buffer, offset, length);
+ }
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
+ validatePositionedReadArgs(position, buffer, offset, length);
+ if (length == 0) {
+ return;
+ }
+ if (inputStream instanceof ExtendedInputStream) {
+ final ByteBuffer buf = ByteBuffer.wrap(buffer, offset, length);
+ try {
+ if (((ExtendedInputStream) inputStream).readFully(position, buf)) {
+ // readFullyStateless returns true once bytesRead > 0, even if the
+ // buffer is only partially filled. Check that the buffer is full.
+ if (buf.hasRemaining()) {
+ throw new EOFException("End of file reached before reading fully.");
+ }
+ return;
+ }
+ } catch (EOFException e) {
+ throw e;
+ }
+ }
+ synchronized (positionedReadLock) {
+ int remaining = length;
+ int off = offset;
+ while (remaining > 0) {
+ int n = readAtPositionSeekRestoreByteArray(position + (length - remaining), buffer, off, remaining);
+ if (n < 0) {
+ throw new EOFException("End of file reached before reading fully.");
+ }
+ off += n;
+ remaining -= n;
+ }
+ }
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer) throws IOException {
+ readFully(position, buffer, 0, buffer.length);
+ }
+
+ private int readAtPositionSeekRestoreByteArray(long position, byte[] buffer, int offset, int length)
+ throws IOException {
+ final long oldPos = getPos();
+ try {
+ ((Seekable) inputStream).seek(position);
+ return read(buffer, offset, length);
+ } catch (EOFException e) {
+ return -1;
+ } finally {
+ ((Seekable) inputStream).seek(oldPos);
+ }
+ }
}
diff --git a/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java b/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
index 86df63949b5..eacf633470d 100644
--- a/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
+++ b/hadoop-ozone/ozonefs-common/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
@@ -17,8 +17,10 @@
package org.apache.hadoop.fs.ozone;
+import static org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper.SOURCE_SIZE;
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.anyString;
import static org.mockito.Mockito.mock;
@@ -28,6 +30,7 @@
import com.google.common.collect.ImmutableList;
import java.io.ByteArrayInputStream;
+import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
@@ -42,15 +45,22 @@
import org.apache.hadoop.crypto.CryptoInputStream;
import org.apache.hadoop.crypto.Decryptor;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Seekable;
import org.apache.hadoop.fs.StreamCapabilities;
+import org.apache.hadoop.hdds.scm.storage.ByteReaderStrategy;
+import org.apache.hadoop.hdds.scm.storage.ExtendedInputStream;
+import org.apache.hadoop.hdds.scm.storage.PositionedReadTestHelper;
import org.apache.hadoop.ozone.client.io.KeyInputStream;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
/**
* Tests for {@link OzoneFSInputStream}.
*/
public class TestOzoneFSInputStream {
+ private static final byte CORRUPT_BYTE = (byte) 0x5A;
+
private static final List> BUFFER_CONSTRUCTORS =
ImmutableList.of(ByteBuffer::allocate, ByteBuffer::allocateDirect);
@@ -187,4 +197,294 @@ public int read() {
};
}
+ @Test
+ public void testByteBufferPositionedReadNegativePositionThrows() throws Exception {
+ // read(long, ByteBuffer) must throw EOFException for negative positions,
+ // aligning with the byte-array PositionedReadable behaviour.
+ final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ final InterleavingSeekableInputStream underlying =
+ new InterleavingSeekableInputStream(source);
+ try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
+ new FileSystem.Statistics("test"))) {
+ ByteBuffer buf = ByteBuffer.allocate(16);
+ assertThrows(EOFException.class, () -> subject.read(-1L, buf));
+ }
+ }
+
+ @Test
+ @Timeout(value = 30)
+ public void testConcurrentPositionedRead() throws Exception {
+ final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ final InterleavingSeekableInputStream underlying =
+ new InterleavingSeekableInputStream(source);
+ try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
+ new FileSystem.Statistics("test"))) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(source,
+ (offset, buf) -> subject.readFully(offset, buf));
+ }
+ }
+
+ @Test
+ @Timeout(value = 30)
+ public void testConcurrentPositionedReadEcFallback() throws Exception {
+ final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ final EcInterleavingInputStream underlying =
+ new EcInterleavingInputStream(source);
+ try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
+ new FileSystem.Statistics("test"))) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(source,
+ (offset, buf) -> subject.readFully(offset, buf));
+ }
+ }
+
+ @Test
+ @Timeout(value = 30)
+ public void testByteArrayFallbackWorksForSeekableOnlyStream() throws Exception {
+ // Regression: the old routing through read(long, ByteBuffer) would cast to
+ // ByteBufferReadable in readAtPositionSeekRestore and throw ClassCastException
+ // for a stream that only implements Seekable (not ByteBufferReadable).
+ final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ final SeekableOnlyInputStream underlying = new SeekableOnlyInputStream(source);
+ try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
+ new FileSystem.Statistics("test"))) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(source, (offset, buf) -> {
+ byte[] arr = new byte[buf.remaining()];
+ subject.readFully(offset, arr);
+ buf.put(arr);
+ });
+ }
+ }
+
+ @Test
+ @Timeout(value = 30)
+ public void testConcurrentByteArrayPositionedReadEcFallback() throws Exception {
+ final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ final EcInterleavingInputStream underlying =
+ new EcInterleavingInputStream(source);
+ try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
+ new FileSystem.Statistics("test"))) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(source, (offset, buf) -> {
+ byte[] arr = new byte[buf.remaining()];
+ subject.readFully(offset, arr);
+ buf.put(arr);
+ });
+ }
+ }
+
+ @Test
+ @Timeout(value = 30)
+ public void testConcurrentMixedApiEcFallback() throws Exception {
+ // ByteBuffer and byte-array callers share positionedReadLock; verify no interleaving.
+ final byte[] source = RandomUtils.secure().randomBytes(SOURCE_SIZE);
+ final EcInterleavingInputStream underlying =
+ new EcInterleavingInputStream(source);
+ try (OzoneFSInputStream subject = new OzoneFSInputStream(underlying,
+ new FileSystem.Statistics("test"))) {
+ PositionedReadTestHelper.runConcurrentPositionedReads(source, (offset, buf) -> {
+ if ((offset & 1) == 0) {
+ subject.readFully(offset, buf);
+ } else {
+ byte[] arr = new byte[buf.remaining()];
+ subject.readFully(offset, arr);
+ buf.put(arr);
+ }
+ });
+ }
+ }
+
+ /**
+ * Mimics KeyInputStream synchronized per-operation seek/read where multi-steps
+ * positioned reads must still be serialized at the FS layer.
+ */
+ private static final class InterleavingSeekableInputStream extends InputStream
+ implements Seekable, org.apache.hadoop.fs.ByteBufferReadable {
+
+ private final InterleavingReadState readState;
+
+ private InterleavingSeekableInputStream(byte[] data) {
+ this.readState = new InterleavingReadState(data);
+ }
+
+ @Override
+ public synchronized void seek(long p) {
+ readState.seek(p);
+ }
+
+ @Override
+ public synchronized long getPos() {
+ return readState.getPos();
+ }
+
+ @Override
+ public synchronized boolean seekToNewSource(long targetPos) {
+ return false;
+ }
+
+ @Override
+ public int read() {
+ return -1;
+ }
+
+ @Override
+ public synchronized int read(ByteBuffer buf) {
+ return readState.read(buf);
+ }
+ }
+
+ /**
+ * Mimics an erasure-coded key stream: {@link ExtendedInputStream#readFully}
+ * returns {@code false}, so {@link OzoneFSInputStream} falls back to
+ * seek-read-restore on the shared cursor. Implements both ByteBuffer and
+ * byte-array reads so both fallback paths can be exercised.
+ */
+ private static final class EcInterleavingInputStream extends ExtendedInputStream {
+
+ private final InterleavingReadState readState;
+
+ private EcInterleavingInputStream(byte[] data) {
+ this.readState = new InterleavingReadState(data);
+ }
+
+ @Override
+ public boolean readFully(long position, ByteBuffer buffer) {
+ return false;
+ }
+
+ @Override
+ protected int readWithStrategy(ByteReaderStrategy strategy) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public synchronized void seek(long p) {
+ readState.seek(p);
+ }
+
+ @Override
+ public synchronized long getPos() {
+ return readState.getPos();
+ }
+
+ @Override
+ public synchronized boolean seekToNewSource(long targetPos) {
+ return false;
+ }
+
+ @Override
+ public synchronized int read(ByteBuffer buf) {
+ return readState.read(buf);
+ }
+
+ @Override
+ public synchronized int read(byte[] b, int off, int len) {
+ return readState.read(b, off, len);
+ }
+
+ @Override
+ public void unbuffer() {
+ return;
+ }
+ }
+
+ /**
+ * A Seekable stream that does NOT implement ByteBufferReadable. Used to verify
+ * that the byte-array positioned-read fallback uses read(byte[]) rather than
+ * casting to ByteBufferReadable (which would throw ClassCastException).
+ */
+ private static final class SeekableOnlyInputStream extends InputStream
+ implements Seekable {
+
+ private final byte[] data;
+ private int pos;
+
+ private SeekableOnlyInputStream(byte[] data) {
+ this.data = data;
+ }
+
+ @Override
+ public synchronized int read() {
+ return pos < data.length ? (data[pos++] & 0xFF) : -1;
+ }
+
+ @Override
+ public synchronized int read(byte[] b, int off, int len) {
+ if (pos >= data.length) {
+ return -1;
+ }
+ int n = Math.min(len, data.length - pos);
+ System.arraycopy(data, pos, b, off, n);
+ pos += n;
+ return n;
+ }
+
+ @Override
+ public synchronized void seek(long newPos) {
+ pos = (int) newPos;
+ }
+
+ @Override
+ public synchronized long getPos() {
+ return pos;
+ }
+
+ @Override
+ public boolean seekToNewSource(long targetPos) {
+ return false;
+ }
+ }
+
+ private static final class InterleavingReadState {
+ private final byte[] data;
+ private long pos;
+ private final ThreadLocal expectedReadPos = new ThreadLocal<>();
+
+ private InterleavingReadState(byte[] data) {
+ this.data = data;
+ }
+
+ private void seek(long p) {
+ pos = p;
+ expectedReadPos.set(p);
+ }
+
+ private long getPos() {
+ return pos;
+ }
+
+ private int read(ByteBuffer buf) {
+ Long expected = expectedReadPos.get();
+ if (expected != null && pos != expected) {
+ int len = buf.remaining();
+ for (int i = 0; i < len; i++) {
+ buf.put(CORRUPT_BYTE);
+ }
+ return len;
+ }
+ int toRead = Math.min(buf.remaining(), data.length - (int) pos);
+ if (toRead <= 0) {
+ return -1;
+ }
+ buf.put(data, (int) pos, toRead);
+ pos += toRead;
+ expectedReadPos.remove();
+ return toRead;
+ }
+
+ private int read(byte[] b, int off, int len) {
+ Long expected = expectedReadPos.get();
+ if (expected != null && pos != expected) {
+ java.util.Arrays.fill(b, off, off + len, CORRUPT_BYTE);
+ return len;
+ }
+ int toRead = Math.min(len, data.length - (int) pos);
+ if (toRead <= 0) {
+ return -1;
+ }
+ System.arraycopy(data, (int) pos, b, off, toRead);
+ pos += toRead;
+ expectedReadPos.remove();
+ return toRead;
+ }
+ }
+
}