This is an automated email from the ASF dual-hosted git repository.
SteNicholas pushed a commit to branch branch-0.6
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.6 by this push:
new 0bf9c5a58 [CELEBORN-2253][0.6] Fix IndexOutOfBoundsException reading
shuffle data from HDFS
0bf9c5a58 is described below
commit 0bf9c5a58598501a970c9fd5d05bf11e599c6e17
Author: Cheng Pan <[email protected]>
AuthorDate: Thu May 21 19:52:38 2026 +0800
[CELEBORN-2253][0.6] Fix IndexOutOfBoundsException reading shuffle data
from HDFS
Backport CELEBORN-2253 (https://github.com/apache/celeborn/pull/3683) to
branch-0.6, code and tests are tuned to adapt branch-0.6 due to miss
CELEBORN-2211 (https://github.com/apache/celeborn/pull/3548)
Assisted-by: OpenCode:mimo-v2.5-pro
---
Original PR description.
### 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 #3697 from pan3793/CELEBORN-2253-0.6.
Authored-by: Cheng Pan <[email protected]>
Signed-off-by: SteNicholas <[email protected]>
---
.../service/deploy/worker/storage/FlushTask.scala | 2 +-
.../deploy/worker/storage/FlushTaskSuite.scala | 69 +++++++++++++++++++++-
2 files changed, 69 insertions(+), 2 deletions(-)
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 35ac91715..1bee1cd7f 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
@@ -107,7 +107,7 @@ private[worker] class HdfsFlushTask(
val hadoopFs = StorageManager.hadoopFs.get(Type.HDFS)
val hdfsStream = hadoopFs.append(path, 256 * 1024)
flush(hdfsStream) {
- hdfsStream.write(convertBufferToBytes(buffer, copyBytes, readableBytes))
+ hdfsStream.write(convertBufferToBytes(buffer, copyBytes, readableBytes),
0, readableBytes)
source.incCounter(WorkerSource.HDFS_FLUSH_COUNT)
source.incCounter(WorkerSource.HDFS_FLUSH_SIZE, readableBytes)
}
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..d91280569 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
@@ -18,17 +18,20 @@
package org.apache.celeborn.service.deploy.worker.storage
import java.io.ByteArrayInputStream
+import java.util.{HashMap => JHashMap}
import io.netty.buffer.{ByteBufAllocator, CompositeByteBuf,
UnpooledByteBufAllocator}
import org.apache.commons.io.IOUtils
+import org.apache.hadoop.fs.{FileSystem, FSDataOutputStream, Path}
import org.mockito.ArgumentCaptor
-import org.mockito.ArgumentMatchersSugar.eqTo
+import org.mockito.ArgumentMatchersSugar.{any, eqTo}
import org.mockito.MockitoSugar.{verify, _}
import org.scalatest.prop.TableDrivenPropertyChecks.forAll
import org.scalatest.prop.Tables.Table
import org.apache.celeborn.CelebornFunSuite
import org.apache.celeborn.common.metrics.source.AbstractSource
+import org.apache.celeborn.common.protocol.StorageInfo.Type
import org.apache.celeborn.server.common.service.mpu.MultipartUploadHandler
import org.apache.celeborn.service.deploy.worker.WorkerSource
@@ -84,6 +87,70 @@ 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
+
+ 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 mockFs = mock[FileSystem]
+ val mockPath = mock[Path]
+
+ val fsMap = new JHashMap[Type, FileSystem]()
+ fsMap.put(Type.HDFS, mockFs)
+ when(mockFs.append(any[Path], any[Int])).thenReturn(mockHdfsStream)
+
+ val origMap = StorageManager.hadoopFs
+ try {
+ StorageManager.hadoopFs = fsMap
+
+ val flushTask = new HdfsFlushTask(
+ mockBuffer,
+ mockPath,
+ mockNotifier,
+ false,
+ mockSource)
+
+ val copyBytesArray = Array.fill[Byte](bufferSize)(0xFF.toByte)
+ flushTask.flush(copyBytesArray)
+
+ 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()
+ } finally {
+ StorageManager.hadoopFs = origMap
+ }
+ }
+ }
+
def runTest(
builder: (
CompositeByteBuf,
