This is an automated email from the ASF dual-hosted git repository.
szetszwo 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 7d9223cd5a9 HDDS-15920. Fix ByteBuffer positioned read behavior at EOF
in OzoneFSInputStream (#10827)
7d9223cd5a9 is described below
commit 7d9223cd5a94a7073f81ac54445c5cb4053df8ec
Author: Chung En Lee <[email protected]>
AuthorDate: Wed Jul 29 01:55:49 2026 +0800
HDDS-15920. Fix ByteBuffer positioned read behavior at EOF in
OzoneFSInputStream (#10827)
---
.../hdds/scm/storage/MultipartInputStream.java | 8 ++
.../hadoop/fs/ozone/TestOzoneFSInputStream.java | 143 ++++++++++++++-------
.../apache/hadoop/fs/ozone/OzoneFSInputStream.java | 8 +-
3 files changed, 109 insertions(+), 50 deletions(-)
diff --git
a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java
index 221a48be828..075ab08fe5a 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
@@ -196,6 +196,10 @@ public boolean readFully(long position, ByteBuffer buffer)
throws IOException {
final long oldPos = getPos();
seek(position);
try {
+ int remainingBeforeRead = buffer.remaining();
+ if (remainingBeforeRead == 0) {
+ return true;
+ }
read(new ByteBufferReader(buffer) {
@Override
int readImpl(InputStream inputStream) throws IOException {
@@ -203,6 +207,10 @@ int readImpl(InputStream inputStream) throws IOException {
.readFully(getBuffer(), false);
}
});
+ if (remainingBeforeRead - buffer.remaining() == 0) {
+ throw new EOFException("EOF encountered at pos: " + position +
+ " for key: " + key);
+ }
} finally {
seek(oldPos);
}
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 cc2f19da2a6..c4073167faa 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
@@ -33,6 +33,7 @@
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.util.UUID;
+import java.util.stream.Stream;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
@@ -40,6 +41,7 @@
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdds.client.DefaultReplicationConfig;
import org.apache.hadoop.hdds.client.ECReplicationConfig;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.StorageType;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.SequenceFile;
@@ -52,9 +54,10 @@
import org.apache.ozone.test.NonHATests;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
import org.junit.jupiter.params.provider.ValueSource;
/**
@@ -64,10 +67,10 @@
public abstract class TestOzoneFSInputStream implements NonHATests.TestCase {
private OzoneClient client;
- private FileSystem fs;
private FileSystem ecFs;
private Path filePath = null;
private byte[] data = null;
+ private String uri = null;
@BeforeAll
void init() throws Exception {
@@ -77,14 +80,15 @@ void init() throws Exception {
OzoneBucket bucket = DataTestUtil.createVolumeAndBucket(client);
// Set the fs.defaultFS and start the filesystem
- String uri = String.format("%s://%s.%s/",
+ uri = String.format("%s://%s.%s/",
OzoneConsts.OZONE_URI_SCHEME, bucket.getName(),
bucket.getVolumeName());
- fs = FileSystem.get(URI.create(uri), cluster().getConf());
- int fileLen = 30 * 1024 * 1024;
- data = string2Bytes(RandomStringUtils.secure().nextAlphanumeric(fileLen));
- filePath = new Path("/" + RandomStringUtils.secure().nextAlphanumeric(5));
- try (FSDataOutputStream stream = fs.create(filePath)) {
- stream.write(data);
+ try (FileSystem fs = FileSystem.get(URI.create(uri),
cluster().getConf());) {
+ int fileLen = 30 * 1024 * 1024;
+ data =
string2Bytes(RandomStringUtils.secure().nextAlphanumeric(fileLen));
+ filePath = new Path("/" +
RandomStringUtils.secure().nextAlphanumeric(5));
+ try (FSDataOutputStream stream = fs.create(filePath)) {
+ stream.write(data);
+ }
}
// create EC bucket to be used by OzoneFileSystem
@@ -106,12 +110,17 @@ void init() throws Exception {
@AfterAll
void shutdown() {
- closeQuietly(client, fs, ecFs);
+ closeQuietly(client, ecFs);
}
- @Test
- public void testO3FSSingleByteRead() throws IOException {
- try (FSDataInputStream inputStream = fs.open(filePath)) {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testO3FSSingleByteRead(boolean isStreamEnable) throws
IOException {
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataInputStream inputStream = fs.open(filePath)) {
byte[] value = new byte[data.length];
int i = 0;
while (true) {
@@ -128,9 +137,13 @@ public void testO3FSSingleByteRead() throws IOException {
}
}
- @Test
- public void testByteBufferPositionedRead() throws IOException {
- try (FSDataInputStream inputStream = fs.open(filePath)) {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testByteBufferPositionedRead(boolean isStreamEnable) throws
IOException {
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataInputStream inputStream = fs.open(filePath)) {
int bufferCapacity = 20;
ByteBuffer buffer = ByteBuffer.allocate(bufferCapacity);
long currentPos = inputStream.getPos();
@@ -182,9 +195,12 @@ public void testByteBufferPositionedRead() throws
IOException {
}
@ParameterizedTest
- @ValueSource(ints = { -1, 30 * 1024 * 1024, 30 * 1024 * 1024 + 1 })
- public void testByteBufferPositionedReadWithInvalidPosition(int position)
throws IOException {
- try (FSDataInputStream inputStream = fs.open(filePath)) {
+ @MethodSource("isStreamEnableAndData")
+ public void testByteBufferPositionedReadWithInvalidPosition(boolean
isStreamEnable, int position) throws IOException {
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataInputStream inputStream = fs.open(filePath)) {
long currentPos = inputStream.getPos();
ByteBuffer buffer = ByteBuffer.allocate(20);
assertEquals(-1, inputStream.read(position, buffer));
@@ -193,9 +209,13 @@ public void
testByteBufferPositionedReadWithInvalidPosition(int position) throws
}
}
- @Test
- public void testByteBufferPositionedReadFully() throws IOException {
- try (FSDataInputStream inputStream = fs.open(filePath)) {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testByteBufferPositionedReadFully(boolean isStreamEnable) throws
IOException {
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataInputStream inputStream = fs.open(filePath)) {
int bufferCapacity = 20;
long currentPos = inputStream.getPos();
ByteBuffer buffer = ByteBuffer.allocate(bufferCapacity);
@@ -235,9 +255,13 @@ public void testByteBufferPositionedReadFully() throws
IOException {
}
@ParameterizedTest
- @ValueSource(ints = { -1, 30 * 1024 * 1024, 30 * 1024 * 1024 + 1 })
- public void testByteBufferPositionedReadFullyWithInvalidPosition(int
position) throws IOException {
- try (FSDataInputStream inputStream = fs.open(filePath)) {
+ @MethodSource("isStreamEnableAndData")
+ public void testByteBufferPositionedReadFullyWithInvalidPosition(
+ boolean isStreamEnable, int position) throws IOException {
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataInputStream inputStream = fs.open(filePath)) {
long currentPos = inputStream.getPos();
ByteBuffer buffer = ByteBuffer.allocate(20);
assertThrows(EOFException.class, () -> inputStream.readFully(position,
buffer));
@@ -246,9 +270,13 @@ public void
testByteBufferPositionedReadFullyWithInvalidPosition(int position) t
}
}
- @Test
- public void testO3FSMultiByteRead() throws IOException {
- try (FSDataInputStream inputStream = fs.open(filePath)) {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testO3FSMultiByteRead(boolean isStreamEnable) throws IOException
{
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataInputStream inputStream = fs.open(filePath)) {
byte[] value = new byte[data.length];
byte[] tmp = new byte[1 * 1024 * 1024];
int i = 0;
@@ -265,10 +293,13 @@ public void testO3FSMultiByteRead() throws IOException {
}
}
- @Test
- public void testO3FSByteBufferRead() throws IOException {
- try (FSDataInputStream inputStream = fs.open(filePath)) {
-
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testO3FSByteBufferRead(boolean isStreamEnable) throws
IOException {
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataInputStream inputStream = fs.open(filePath)) {
ByteBuffer buffer = ByteBuffer.allocate(1024 * 1024);
int byteRead = inputStream.read(buffer);
@@ -281,29 +312,34 @@ public void testO3FSByteBufferRead() throws IOException {
}
}
- @Test
- public void testSequenceFileReaderSync() throws IOException {
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testSequenceFileReaderSync(boolean isStreamEnable) throws
IOException {
File srcfile = new File("src/test/resources/testSequenceFile");
Path path = new Path("/" + RandomStringUtils.secure().nextAlphanumeric(5));
InputStream input = new
BufferedInputStream(Files.newInputStream(srcfile.toPath()));
// Upload test SequenceFile file
- FSDataOutputStream output = fs.create(path);
- IOUtils.copyBytes(input, output, 4096, true);
- input.close();
-
- // Start SequenceFile.Reader test
- SequenceFile.Reader in = new SequenceFile.Reader(fs, path,
cluster().getConf());
- long blockStart = -1;
- // EOFException should not occur.
- in.sync(0);
- blockStart = in.getPosition();
- // The behavior should be consistent with HDFS
- assertEquals(srcfile.length(), blockStart);
- in.close();
+ OzoneConfiguration conf = cluster().getConf();
+ conf.setBoolean("ozone.client.stream.readblock.enable", isStreamEnable);
+ try (FileSystem fs = FileSystem.get(URI.create(uri), conf);
+ FSDataOutputStream output = fs.create(path);) {
+ IOUtils.copyBytes(input, output, 4096, true);
+ input.close();
+
+ // Start SequenceFile.Reader test
+ SequenceFile.Reader in = new SequenceFile.Reader(fs, path,
cluster().getConf());
+ long blockStart = -1;
+ // EOFException should not occur.
+ in.sync(0);
+ blockStart = in.getPosition();
+ // The behavior should be consistent with HDFS
+ assertEquals(srcfile.length(), blockStart);
+ }
}
- @Test
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
public void testSequenceFileReaderSyncEC() throws IOException {
File srcfile = new File("src/test/resources/testSequenceFile");
Path path = new Path("/" + RandomStringUtils.secure().nextAlphanumeric(5));
@@ -324,4 +360,15 @@ public void testSequenceFileReaderSyncEC() throws
IOException {
assertEquals(srcfile.length(), blockStart);
in.close();
}
+
+ static Stream<Arguments> isStreamEnableAndData() {
+ return Stream.of(
+ Arguments.of(false, -1),
+ Arguments.of(false, 30 * 1024 * 1024),
+ Arguments.of(false, 30 * 1024 * 1024 + 1),
+ Arguments.of(true, -1),
+ Arguments.of(true, 30 * 1024 * 1024),
+ Arguments.of(true, 30 * 1024 * 1024 + 1)
+ );
+ }
}
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 e640c1e6d17..a9c2c8b2f0f 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,8 +171,12 @@ public int read(long position, ByteBuffer buf) throws
IOException {
}
if (inputStream instanceof ExtendedInputStream) {
final int remainingBeforeRead = buf.remaining();
- if (((ExtendedInputStream) inputStream).readFully(position, buf)) {
- return remainingBeforeRead - buf.remaining();
+ try {
+ if (((ExtendedInputStream) inputStream).readFully(position, buf)) {
+ return remainingBeforeRead - buf.remaining();
+ }
+ } catch (EOFException e) {
+ return -1;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
