[CELEBORN-1917] Support celeborn.client.push.maxBytesSizeInFlight by DDDominik · Pull Request #3248 · apache/celeborn

DDDominik · 2025-05-08T11:56:46Z

What changes were proposed in this pull request?

add data size limitation to inflight data by introducing a new configuration: celeborn.client.push.maxBytesInFlight.perWorker/total and defaults to celeborn.client.push.buffer.max.size * celeborn.client.push.maxReqsInFlight.perWorker/total.
for backward compatibility, also add a control: celeborn.client.push.maxReqsInFlight.enabled.

Why are the changes needed?

celeborn do supports limiting the number of push inflight requests via celeborn.client.push.maxReqsInFlight.perWorker/total. this is a good constraint to memory usage where most requests do not exceed celeborn.client.push.buffer.max.size. however, in a vectorized shuffle (like blaze and gluten), a request might be greatly larger then the max buffer size, leading to too much inflight data and results OOM.

Does this PR introduce any user-facing change?

Yes, add new config for client

How was this patch tested?

test on local env

mridulm · 2025-05-09T19:37:59Z

+CC @venkata91, @rmcyang

SteNicholas · 2025-05-13T12:37:57Z

@DDDominik, any update?

turboFei · 2025-06-02T11:15:30Z

Please update the docs with below command:

    UPDATE=1 build/mvn clean test -pl common -am -Dtest=none -DwildcardSuites=org.apache.celeborn.ConfigurationSuite

turboFei · 2025-06-02T14:52:22Z

Have not see this error before, it should be related to this PR.
https://github.com/apache/celeborn/actions/runs/15391457214/job/43306799162?pr=3248

[info] - celeborn spark integration test - resubmit a failed barrier stage across jobs *** FAILED ***
[info]   org.apache.spark.SparkException: Job aborted due to stage failure: ResultStage 5 (collect at CelebornFetchFailureSuite.scala:379) has failed the maximum allowable number of times: 1. Most recent failure reason:
[info] org.apache.spark.shuffle.FetchFailedException: Celeborn FetchFailure appShuffleId/shuffleId: 0/0
[info] 	at org.apache.spark.shuffle.celeborn.CelebornShuffleReader.read(CelebornShuffleReader.scala:96)
[info] 	at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:106)
[info] 	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
[info] 	at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
[info] 	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
[info] 	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:171)
[info] 	at org.apache.spark.scheduler.Task.run(Task.scala:147)
[info] 	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$5(Executor.scala:647)
[info] 	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:80)
[info] 	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:77)
[info] 	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:99)
[info] 	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:650)
[info] 	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
[info] 	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
[info] 	at java.base/java.lang.Thread.run(Thread.java:840)
[info] Caused by: org.apache.celeborn.common.exception.CelebornRuntimeException: Get invalid shuffle id -1
[info] 	at org.apache.spark.shuffle.celeborn.SparkUtils.celebornShuffleId(SparkUtils.java:146)
[info] 	at org.apache.spark.shuffle.celeborn.CelebornShuffleReader.read(CelebornShuffleReader.scala:84)
[info] 	... 14 more

https://github.com/apache/celeborn/actions/runs/15391457221/job/43306794218?pr=3248

- celeborn spark integration test - resubmit a failed barrier stage across jobs *** FAILED ***
  org.apache.spark.SparkException: Job aborted due to stage failure: ResultStage 5 (collect at CelebornFetchFailureSuite.scala:379) has failed the maximum allowable number of times: 1. Most recent failure reason:
org.apache.spark.shuffle.FetchFailedException: Celeborn FetchFailure appShuffleId/shuffleId: 0/0
	at org.apache.spark.shuffle.celeborn.CelebornShuffleReader.read(CelebornShuffleReader.scala:96)
	at org.apache.spark.rdd.ShuffledRDD.compute(ShuffledRDD.scala:106)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:374)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:338)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:93)
	at org.apache.spark.TaskContext.runTaskWithListeners(TaskContext.scala:171)
	at org.apache.spark.scheduler.Task.run(Task.scala:147)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$5(Executor.scala:647)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally(SparkErrorUtils.scala:80)
	at org.apache.spark.util.SparkErrorUtils.tryWithSafeFinally$(SparkErrorUtils.scala:77)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:99)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:650)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)
Caused by: org.apache.celeborn.common.exception.CelebornRuntimeException: Get invalid shuffle id -1
	at org.apache.spark.shuffle.celeborn.SparkUtils.celebornShuffleId(SparkUtils.java:146)
	at org.apache.spark.shuffle.celeborn.CelebornShuffleReader.read(CelebornShuffleReader.scala:84)
	... 14 more

FMX

This RP can be helpful when there is memory pressure on executors.

github-actions · 2025-07-03T08:37:39Z

This PR is stale because it has been open 20 days with no activity. Remove stale label or comment or this will be closed in 10 days.

SteNicholas · 2025-07-21T11:18:25Z

Ping @FMX, @RexXiong, @mridulm.

…kup - Modified reader logic to use latest shuffle ID when no valid finished stages available - Fixes barrier stage resubmission across jobs test failure

DDDominik

LGTM.

codecov · 2025-07-22T07:20:44Z

Codecov Report

Attention: Patch coverage is 29.00000% with 71 lines in your changes missing coverage. Please review.

Project coverage is 63.56%. Comparing base (8ae9737) to head (a46773d).
Report is 18 commits behind head on main.

Files with missing lines	Patch %	Lines
.../celeborn/common/write/InFlightRequestTracker.java	11.27%	56 Missing and 7 partials ⚠️
...cala/org/apache/celeborn/common/CelebornConf.scala	71.43%	8 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3248      +/-   ##
==========================================
- Coverage   63.57%   63.56%   -0.00%     
==========================================
  Files         348      351       +3     
  Lines       21300    21571     +271     
  Branches     1879     1914      +35     
==========================================
+ Hits        13539    13710     +171     
- Misses       6781     6861      +80     
- Partials      980     1000      +20

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

RexXiong

LGTM, only a nit

mridulm

Just one query, rest looks good - thanks !

mridulm · 2025-07-22T14:44:07Z

+              || (maxInFlightBytesSizeEnabled
+                  && totalInflightBytes.sum() <= maxInFlightBytesSizeTotal
+                  && batchBytesSize.sum() <= maxInFlightBytesSizePerWorker)) {


My earlier query was, should this be || ?
As in, either total inflight is high, or inflight to a specific worker is high (assuming per worker threshold is lower than total !)

Suggested change

|| (maxInFlightBytesSizeEnabled

&& totalInflightBytes.sum() <= maxInFlightBytesSizeTotal

&& batchBytesSize.sum() <= maxInFlightBytesSizePerWorker)) {

|| (maxInFlightBytesSizeEnabled && (

totalInflightBytes.sum() <= maxInFlightBytesSizeTotal ||

batchBytesSize.sum() <= maxInFlightBytesSizePerWorker))) {

Looks like the PR got merged before this was addressed.
If this is not a concern, that should be fine - else let us do a follow up.

To avoid this comment being ignored, I will submit a PR first

FMX

LGTM. Thanks. Merged into main(v0.7.0).

turboFei · 2025-07-22T17:27:49Z

+  val CLIENT_PUSH_MAX_BYTES_SIZE_IN_FLIGHT_PERWORKER: OptionalConfigEntry[Long] =
+    buildConf("celeborn.client.push.maxBytesSizeInFlight.perWorker")
+      .categories("client")
+      .version("0.6.1")


merging to 0.6.1 as well

### What changes were proposed in this pull request? add data size limitation to inflight data by introducing a new configuration: `celeborn.client.push.maxBytesInFlight.perWorker/total` and defaults to `celeborn.client.push.buffer.max.size * celeborn.client.push.maxReqsInFlight.perWorker/total`. for backward compatibility, also add a control: `celeborn.client.push.maxReqsInFlight.enabled`. ### Why are the changes needed? celeborn do supports limiting the number of push inflight requests via `celeborn.client.push.maxReqsInFlight.perWorker/total`. this is a good constraint to memory usage where most requests do not exceed `celeborn.client.push.buffer.max.size`. however, in a vectorized shuffle (like blaze and gluten), a request might be greatly larger then the max buffer size, leading to too much inflight data and results OOM. ### Does this PR introduce _any_ user-facing change? Yes, add new config for client ### How was this patch tested? test on local env Closes #3248 from DDDominik/CELEBORN-1917. Lead-authored-by: DDDominik <1015545832@qq.com> Co-authored-by: SteNicholas <programgeek@163.com> Co-authored-by: DDDominik <zhuangxian@kuaishou.com> Signed-off-by: mingji <fengmingxiao.fmx@alibaba-inc.com> (cherry picked from commit 0ed590d) Signed-off-by: Wang, Fei <fwang12@ebay.com>

turboFei · 2025-07-22T17:29:19Z

Merged into branch-0.6(0.6.1)

github-actions Bot added module:client module:common labels May 8, 2025

SteNicholas reviewed May 9, 2025

View reviewed changes

Comment thread common/src/main/java/org/apache/celeborn/common/write/PushState.java

SteNicholas force-pushed the main branch 2 times, most recently from 5590ef0 to 0dffcf6 Compare May 26, 2025 09:57

github-actions Bot added the module:flink label May 27, 2025

github-actions Bot added the kind:documentation label Jun 2, 2025

FMX reviewed Jun 12, 2025

View reviewed changes

github-actions Bot added the stale label Jul 3, 2025

FMX removed the stale label Jul 10, 2025

SteNicholas changed the title ~~[CELEBORN-1917] supports celeborn.client.push.maxBytesSizeInFlight~~ [CELEBORN-1917] Support celeborn.client.push.maxBytesSizeInFlight Jul 21, 2025

SteNicholas force-pushed the CELEBORN-1917 branch from b906d44 to 8e67ae6 Compare July 21, 2025 09:34

SteNicholas requested a review from FMX July 21, 2025 11:20

SteNicholas force-pushed the main branch 2 times, most recently from e8622e1 to d09b424 Compare July 21, 2025 11:38

DDDominik and others added 9 commits July 21, 2025 20:03

[CELEBORN-1917] supports celeborn.client.push.maxBytesSizeInFlight

9deade2

update the invoker in FlinkShuffleClientImpl

d9bc4e6

update client.md

6011bd3

Fix InFlightRequestTracker batch bytes tracking inconsistency

675c300

Fix violations with mvn spotless:apply

b2ecb86

[CELEBORN-1917][CLIENT] Fix barrier stage resubmission shuffle ID loo…

aefb85c

…kup - Modified reader logic to use latest shuffle ID when no valid finished stages available - Fixes barrier stage resubmission across jobs test failure

Fix violations with mvn spotless:apply

9cc9e27

fix empty shuffleId

a1881ed

[CELEBORN-1917] Support celeborn.client.push.maxBytesSizeInFlight

0a3f223

SteNicholas force-pushed the CELEBORN-1917 branch from 8e67ae6 to 0a3f223 Compare July 21, 2025 12:04

DDDominik commented Jul 21, 2025

View reviewed changes

mridulm reviewed Jul 21, 2025

View reviewed changes

SteNicholas requested a review from mridulm July 22, 2025 06:18

DDDominik commented Jul 22, 2025

View reviewed changes

Comment thread common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala

Comment thread common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala

[CELEBORN-1917] Support celeborn.client.push.maxBytesSizeInFlight

a46773d

SteNicholas force-pushed the CELEBORN-1917 branch from 5401626 to a46773d Compare July 22, 2025 07:02

RexXiong approved these changes Jul 22, 2025

View reviewed changes

Comment thread common/src/main/java/org/apache/celeborn/common/write/InFlightRequestTracker.java Outdated

[CELEBORN-1917] Support celeborn.client.push.maxBytesSizeInFlight

3efc818

mridulm reviewed Jul 22, 2025

View reviewed changes

FMX approved these changes Jul 22, 2025

View reviewed changes

FMX closed this in 0ed590d Jul 22, 2025

turboFei reviewed Jul 22, 2025

View reviewed changes

turboFei mentioned this pull request Jul 24, 2025

[CELEBORN-1917][FOLLOWUP] Do not wait if total inflight or inflight to a specific worker is lower than threshold #3386

Closed

Conversation

DDDominik commented May 8, 2025

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

Uh oh!

mridulm commented May 9, 2025

Uh oh!

SteNicholas commented May 13, 2025

Uh oh!

turboFei commented Jun 2, 2025

Uh oh!

turboFei commented Jun 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

FMX left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

github-actions Bot commented Jul 3, 2025

Uh oh!

SteNicholas commented Jul 21, 2025

Uh oh!

DDDominik left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

codecov Bot commented Jul 22, 2025

Codecov Report

Uh oh!

RexXiong left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

mridulm left a comment

Choose a reason for hiding this comment

Uh oh!

mridulm Jul 22, 2025

Choose a reason for hiding this comment

Uh oh!

mridulm Jul 22, 2025

Choose a reason for hiding this comment

Uh oh!

turboFei Jul 24, 2025

Choose a reason for hiding this comment

Uh oh!

FMX left a comment

Choose a reason for hiding this comment

Uh oh!

turboFei Jul 22, 2025

Choose a reason for hiding this comment

Uh oh!

turboFei commented Jul 22, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

turboFei commented Jun 2, 2025 •

edited

Loading

FMX left a comment •

edited

Loading