[CELEBORN-2253] Fix IndexOutOfBoundsException reading shuffle data from HDFS

[1fanwang]

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.

[Copilot]

Pull request overview

Fixes potential shuffle data corruption in the HDFS flush path by ensuring only the valid portion of a reused buffer is written to HDFS, aligning behavior with the previously-fixed S3/OSS flush implementations.

Changes:

• Update HdfsFlushTask to call FSDataOutputStream.write(bytes, 0, size) instead of writing the full array length.
• Add a unit test covering HDFS flush behavior when the provider’s reusable buffer is equal to, larger than, or smaller than the payload size.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTask.scala	Fixes HDFS write to respect readableBytes length and prevent trailing garbage bytes from being persisted.
worker/src/test/scala/org/apache/celeborn/service/deploy/worker/storage/FlushTaskSuite.scala	Adds regression coverage to validate HDFS flush uses the correct offset/length when writing.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[SteNicholas]

Thanks. Merged to main(v0.7.0).