You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by ad...@apache.org on 2026/09/22 17:49:13 UTC
(ozone) branch master updated: HDDS-16400. Fix thread-safety of positioned reads in OzoneFSInputStream (#11245)
This is an automated email from the ASF dual-hosted git repository.
adoroszlai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new f73fa1a506a HDDS-16400. Fix thread-safety of positioned reads in OzoneFSInputStream (#11245)
f73fa1a506a is described below
commit f73fa1a506a304a39dd91d3e53b3d9b3d7c1ff08
Author: Tak Lon (Stephen) Wu <ta...@apache.org>
AuthorDate: Tue Sep 22 10:49:07 2026 -0700
HDDS-16400. Fix thread-safety of positioned reads in OzoneFSInputStream (#11245)
Co-authored-by: Cursor <cu...@cursor.com>
---
.../hadoop/fs/ozone/TestOzoneFSInputStream.java | 8 +-
hadoop-ozone/ozonefs-common/pom.xml | 6 +
.../apache/hadoop/fs/ozone/OzoneFSInputStream.java | 179 +++++++++---
.../hadoop/fs/ozone/TestOzoneFSInputStream.java | 299 +++++++++++++++++++++
4 files changed, 458 insertions(+), 34 deletions(-)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
index c4073167faa..e704f289770 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java
@@ -203,7 +203,13 @@ public void testByteBufferPositionedReadWithInvalidPosition(boolean isStreamEnab
FSDataInputStream inputStream = fs.open(filePath)) {
long currentPos = inputStream.getPos();
ByteBuffer buffer = ByteBuffer.allocate(20);
- assertEquals(-1, inputStream.read(position, buffer));
+ if (position < 0) {
+ // Negative position must throw EOFException (aligned with the byte-array PositionedReadable contract).
+ assertThrows(EOFException.class, () -> inputStream.read(position, buffer));
+ } else {
+ // Position at or past EOF must return -1.
+ assertEquals(-1, inputStream.read(position, buffer));
+ }
// File position should not be changed
assertEquals(currentPos, inputStream.getPos());
}
diff --git a/hadoop-ozone/ozonefs-common/pom.xml b/hadoop-ozone/ozonefs-common/pom.xml
index fe87bf574e8..e08b43268e8 100644
--- a/hadoop-ozone/ozonefs-common/pom.xml
+++ b/hadoop-ozone/ozonefs-common/pom.xml
@@ -94,6 +94,12 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.ozone</groupId>
+ <artifactId>hdds-client</artifactId>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
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..1b8e5e9c9db 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
@@ -35,9 +35,17 @@
/**
* The input stream for Ozone file system.
- *
- * TODO: Make inputStream generic for both rest and rpc clients
- * This class is not thread safe.
+ * <p>
+ * Sequential reads are NOT thread safe.
+ * <p>
+ * Positioned reads are thread safe.
+ * When the underlying stream is an {@link ExtendedInputStream} and
+ * when it supports {@link ExtendedInputStream#readFully(long, ByteBuffer)},
+ * they delegate to it.
+ * Otherwise, they fall back to the default synchronized seek-read-restore implementation.
+ * <p>
+ * Applications must use either sequential reads or position reads at any given time,
+ * but not concurrent sequential/position reads
*/
@InterfaceAudience.Private
@InterfaceStability.Evolving
@@ -46,6 +54,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;
@@ -158,57 +167,161 @@ public void unbuffer() {
}
/**
- * @param buf the ByteBuffer to receive the results of the read operation.
+ * @param buffer to receive the results of the read operation.
* @param position offset
* @return the number of bytes read, possibly zero, or -1 if
* reach end-of-stream
* @throws IOException if there is some error performing the read
*/
@Override
- public int read(long position, ByteBuffer buf) throws IOException {
- if (!buf.hasRemaining()) {
+ public int read(long position, ByteBuffer buffer) throws IOException {
+ if (!buffer.hasRemaining()) {
return 0;
}
- if (inputStream instanceof ExtendedInputStream) {
- final int remainingBeforeRead = buf.remaining();
- try {
- if (((ExtendedInputStream) inputStream).readFully(position, buf)) {
- return remainingBeforeRead - buf.remaining();
- }
- } catch (EOFException e) {
- return -1;
+ if (position < 0) {
+ throw new EOFException("position is negative: " + position);
+ }
+ return readImpl(position, buffer);
+ }
+
+ @Override
+ public int read(long position, byte[] array, int offset, int length) throws IOException {
+ if (length == 0) {
+ return 0;
+ }
+ validatePositionedReadArgs(position, array, offset, length);
+ return readImpl(position, ByteBuffer.wrap(array, offset, length));
+ }
+
+ private int readImpl(long position, ByteBuffer buffer) throws IOException {
+ final Integer n = bestEffortExtendedInputStreamRead(position, buffer, false);
+ if (n != null) {
+ return n;
+ }
+ // Fallback: stateful seek-read-restore on the shared cursor.
+ return bestEffortReadFullySynchronized(position, buffer);
+ }
+
+ /**
+ * @param buffer to receive the results of the read operation.
+ * @param position offset
+ * @throws IOException if there is some error performing the read
+ * @throws EOFException if end of file reached before reading fully
+ */
+ @Override
+ public void readFully(long position, ByteBuffer buffer) throws IOException {
+ if (!buffer.hasRemaining()) {
+ return;
+ }
+ if (position < 0) {
+ throw new EOFException("position is negative: " + position);
+ }
+ readFullyImpl(position, buffer);
+ }
+
+ @Override
+ public void readFully(long position, byte[] array, int offset, int length) throws IOException {
+ if (length == 0) {
+ return;
+ }
+ validatePositionedReadArgs(position, array, offset, length);
+ readFullyImpl(position, ByteBuffer.wrap(array, offset, length));
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer) throws IOException {
+ readFully(position, buffer, 0, buffer.length);
+ }
+
+ private void readFullyImpl(long position, ByteBuffer buffer) throws IOException {
+ final int length = buffer.remaining();
+ final Integer n = bestEffortExtendedInputStreamRead(position, buffer, true);
+ if (n != null) {
+ if (n < length) {
+ throw new EOFException("Failed to read fully length " + length + " at position " + position);
}
+ return;
}
+ final int m = bestEffortReadFullySynchronized(position, buffer);
+ if (m < length) {
+ throw new EOFException("Failed to read fully length " + length + " at position " + position);
+ }
+ }
- long oldPos = this.getPos();
- int bytesRead;
+ /**
+ * Best effort read via {@link ExtendedInputStream#readFully(long, ByteBuffer)}.
+ *
+ * @return the number of bytes read; or null when the native stateless path is unsupported.
+ */
+ private Integer bestEffortExtendedInputStreamRead(long position, ByteBuffer buffer,
+ boolean isReadFully) throws IOException {
+ if (!(inputStream instanceof ExtendedInputStream)) {
+ return null; // not an ExtendedInputStream
+ }
+ final int remainingBeforeRead = buffer.remaining();
try {
- ((Seekable) inputStream).seek(position);
- bytesRead = ((ByteBufferReadable) inputStream).read(buf);
+ if (!((ExtendedInputStream) inputStream).readFully(position, buffer)) {
+ return null; // ExtendedInputStream does not support readFully
+ }
} catch (EOFException e) {
- // Either position is negative or it has reached EOF
+ if (isReadFully) {
+ throw e;
+ }
+ // read() semantics: EOF -> -1 (fall through to bytesRead == 0 check below)
+ }
+ final int bytesRead = remainingBeforeRead - buffer.remaining();
+ if (bytesRead == 0) {
return -1;
- } finally {
- ((Seekable) inputStream).seek(oldPos);
+ }
+ if (statistics != null) {
+ statistics.incrementBytesRead(bytesRead);
}
return bytesRead;
}
/**
- * @param buf the ByteBuffer to receive the results of the read operation.
- * @param position offset
- * @throws IOException if there is some error performing the read
- * @throws EOFException if end of file reached before reading fully
+ * Fallback positioned read via synchronized seek-read-restore.
+ * Uses {@link #read(ByteBuffer)} -- which handles streams that do not implement
+ * {@link org.apache.hadoop.fs.ByteBufferReadable} (e.g. GDPR
+ * {@code javax.crypto.CipherInputStream}) -- so this works for any {@link Seekable}
+ * inner stream.
+ *
+ * @return the number of bytes read
*/
- @Override
- public void readFully(long position, ByteBuffer buf) throws IOException {
- int bytesRead;
- for (int readCount = 0; buf.hasRemaining(); readCount += bytesRead) {
- bytesRead = this.read(position + (long)readCount, buf);
- if (bytesRead < 0) {
- // Still buffer has space to read but stream has already reached EOF
- throw new EOFException("End of file reached before reading fully.");
+ private int bestEffortReadFullySynchronized(long position, ByteBuffer buffer) throws IOException {
+ synchronized (positionedReadLock) {
+ final long oldPos = getPos();
+ try {
+ ((Seekable) inputStream).seek(position);
+ final int n = bestEffortRead(buffer);
+ if (statistics != null) {
+ statistics.incrementBytesRead(n);
+ }
+ return n;
+ } finally {
+ ((Seekable) inputStream).seek(oldPos);
}
}
}
+
+ /**
+ * Best effort read to fill up the buffer using {@link #read(ByteBuffer)}.
+ *
+ * @return the number of bytes read
+ */
+ private int bestEffortRead(ByteBuffer buffer) throws IOException {
+ int readLength = 0;
+ try {
+ while (buffer.hasRemaining()) {
+ final int n = read(buffer);
+ if (n < 0) {
+ return readLength;
+ }
+ readLength += n;
+ }
+ return readLength;
+ } catch (EOFException e) {
+ return readLength;
+ }
+ }
}
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..6954547e739 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<IntFunction<ByteBuffer>> BUFFER_CONSTRUCTORS =
ImmutableList.of(ByteBuffer::allocate, ByteBuffer::allocateDirect);
@@ -187,4 +197,293 @@ 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 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 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 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.
+ */
+ private static final class EcInterleavingInputStream extends ExtendedInputStream {
+
+ private final InterleavingReadState readState;
+
+ private EcInterleavingInputStream(byte[] data) {
+ this.readState = new InterleavingReadState(data);
+ }
+
+ @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 int available() {
+ return data.length - pos;
+ }
+
+ @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<Long> 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;
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org