This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new e329b16ff [CELEBORN-2253] Fix IndexOutOfBoundsException reading
shuffle data from HDFS
e329b16ff is described below
commit e329b16ff70b13419e862acf2dcd7bc05d829ab6
Author: 1fanwang <[email protected]>
AuthorDate: Wed May 13 14:45:53 2026 +0800
[CELEBORN-2253] Fix IndexOutOfBoundsException reading shuffle data from HDFS
### What changes were proposed in this pull request?
`HdfsFlushTask.writeAndRecordMetrics` calls `hdfsStream.write(bytes)`,
which writes the full `bytes.length`. When the provider passes a reusable
`copyBytes` buffer (whose length is `>= size`), this leaks trailing bytes from
previous flushes into the current partition file. Pass the actual readable size
to write only `size` bytes.
### Why are the changes needed?
The S3 and OSS flush paths had the same bug and were fixed in #3600 for
CELEBORN-2263; the HDFS path was missed. Without the fix, shuffle data flushed
to HDFS can be corrupted when `copyBytes` is reused across flushes, and readers
later fail with `IndexOutOfBoundsException` in
`CelebornInputStream.fillBuffer`, for example:
```
IndexOutOfBoundsException: readerIndex(4154253) + length(808530018)
exceeds writerIndex(12457470)
```
### Does this PR resolve a correctness bug?
Yes.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit test in `FlushTaskSuite` mirrors the S3/OSS coverage added in
#3600. It drives `HdfsFlushTask.flush` with `copyBytes` arrays of three sizes
(equal, larger, smaller than the buffer payload), captures the
`FSDataOutputStream.write` arguments via Mockito's `ArgumentCaptor`, and
asserts the offset/length pair matches the buffer content. The test fails on
master with `ArgumentsAreDifferent` at `FlushTask.scala:128` and passes with
the fix.
Closes #3683 from 1fanwang/CELEBORN-2253-fix-hdfs-flush-trailing-bytes.
Authored-by: 1fanwang <[email protected]>
Signed-off-by: SteNicholas <[email protected]>
---
.../service/deploy/worker/storage/FlushTask.scala | 2 +-
.../deploy/worker/storage/FlushTaskSuite.scala | 59 ++++++++++++++++++++++
2 files changed, 60 insertions(+), 1 deletion(-)
diff --git
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala
index a0dd85717..67f45e28b 100644
---
a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala
+++
b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala
@@ -125,7 +125,7 @@ private[worker] class HdfsFlushTask(
hdfsStream: FSDataOutputStream,
bytes: Array[Byte],
size: Int): Unit = {
- hdfsStream.write(bytes)
+ hdfsStream.write(bytes, 0, size)
source.incCounter(WorkerSource.HDFS_FLUSH_COUNT)
source.incCounter(WorkerSource.HDFS_FLUSH_SIZE, size)
}
diff --git
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTaskSuite.scala
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTaskSuite.scala
index da7e2b456..542a142d0 100644
---
a/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTaskSuite.scala
+++
b/worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTaskSuite.scala
@@ -21,6 +21,7 @@ import java.io.ByteArrayInputStream
import io.netty.buffer.{ByteBufAllocator, CompositeByteBuf,
UnpooledByteBufAllocator}
import org.apache.commons.io.IOUtils
+import org.apache.hadoop.fs.{FSDataOutputStream, Path}
import org.mockito.ArgumentCaptor
import org.mockito.ArgumentMatchersSugar.eqTo
import org.mockito.MockitoSugar.{verify, _}
@@ -84,6 +85,64 @@ class FlushTaskSuite extends CelebornFunSuite {
})
}
+ test("HdfsFlushTask flush should work with buffers of various sizes") {
+ val bytes = "another test data".getBytes("UTF-8")
+ val len = bytes.length
+
+ // Define the scenarios: (scenario name, size to allocate)
+ val scenarios = Table(
+ ("description", "allocatedSize"),
+ ("provider buffer is the same size as the buffer", len),
+ ("provider buffer is bigger", len + 10),
+ ("provider buffer smaller", len - 5))
+
+ forAll(scenarios) { (description, bufferSize) =>
+ val mockBuffer = spy(ALLOCATOR.compositeBuffer())
+ mockBuffer.writeBytes(bytes)
+ val mockNotifier = mock[FlushNotifier]
+ val mockSource = mock[AbstractSource]
+ val mockHdfsStream = mock[FSDataOutputStream]
+ val mockPath = mock[Path]
+
+ val flushTask = new HdfsFlushTask(
+ mockBuffer,
+ mockHdfsStream,
+ mockPath,
+ mockNotifier,
+ false, // keepBuffer
+ mockSource)
+
+ // Pre-fill the provider buffer with a sentinel so that the buggy path
+ // (writing the full array instead of size bytes) is detectable: any
+ // trailing bytes that leak through would still be the sentinel.
+ val copyBytesArray = Array.fill[Byte](bufferSize)(0xFF.toByte)
+ flushTask.flush(copyBytesArray)
+
+ // buffer position is not moved
+ assert(mockBuffer.readableBytes() == bytes.length)
+
+ val bytesCaptor = ArgumentCaptor.forClass(classOf[Array[Byte]])
+ val offsetCaptor = ArgumentCaptor.forClass(classOf[Integer])
+ val lengthCaptor = ArgumentCaptor.forClass(classOf[Integer])
+ verify(mockHdfsStream).write(
+ bytesCaptor.capture(),
+ offsetCaptor.capture(),
+ lengthCaptor.capture())
+ verify(mockSource).incCounter(WorkerSource.HDFS_FLUSH_COUNT)
+ verify(mockSource).incCounter(WorkerSource.HDFS_FLUSH_SIZE, bytes.length)
+
+ assert(offsetCaptor.getValue == 0, s"Offset mismatch on: $description")
+ assert(
+ lengthCaptor.getValue == bytes.length,
+ s"Length mismatch on: $description")
+ val capturedBytes = bytesCaptor.getValue
+ .slice(offsetCaptor.getValue, offsetCaptor.getValue +
lengthCaptor.getValue)
+ assert(capturedBytes sameElements bytes, s"Content mismatch on:
$description")
+
+ mockBuffer.release()
+ }
+ }
+
def runTest(
builder: (
CompositeByteBuf,
