This is an automated email from the ASF dual-hosted git repository.
zaynt4606 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 f4d6851663 [CELEBORN-2483] Fix flaky RemoteShuffleMasterSuiteJ
shuffleCount assertions racing with the initial app heartbeat
f4d6851663 is described below
commit f4d6851663d865907070059077eff260d638246f
Author: afterincomparableyum
<[email protected]>
AuthorDate: Mon Oct 19 11:57:28 2026 +0800
[CELEBORN-2483] Fix flaky RemoteShuffleMasterSuiteJ shuffleCount assertions
racing with the initial app heartbeat
### What changes were proposed in this pull request?
- Add the internal conf
`celeborn.test.client.application.heartbeat.enabled` (default `true`). When it
is `false`, `ApplicationHeartbeater` does not schedule the app heartbeat.
- Disable the app heartbeat in `RemoteShuffleMasterSuiteJ`, replacing the
`heartbeatInterval=1h` workaround from CELEBORN-2431.
### Why are the changes needed?
`RemoteShuffleMasterSuiteJ.testRegisterPartitionWithProducer` is still
flaky after CELEBORN-2431:
```
RemoteShuffleMasterSuiteJ.testRegisterPartitionWithProducer:127
expected:<1> but was:<0>
RemoteShuffleMasterSuiteJ.testRegisterPartitionWithProducer:158
expected:<3> but was:<1>
```
`ApplicationHeartbeater` schedules its first heartbeat with an initial
delay of 0. Each heartbeat drains `LifecycleManager#shuffleCount` and the
fallback counts via `sumThenReset`. The `LifecycleManager` is created in
`registerJob`, so the first heartbeat runs at the same time as the test's
`registerPartitionWithProducer` calls and can reset the counter between an
increment and its assertion. Raising the interval to 1h only removed the
periodic heartbeats, not the initial one. The sui [...]
Example failures: [flink1 (11,
1.18)](https://github.com/apache/celeborn/actions/runs/38396707127/job/115293644538),
[flink2 (17,
2.2)](https://github.com/apache/celeborn/actions/runs/38396707127/job/115293645021).
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
`RemoteShuffleMasterSuiteJV120`: 7 tests passed. The `AskSync
HeartbeatFromApplication failed` errors, which appeared once per test in CI,
should no longer appear.
Closes #3875 from
afterincomparableyum/flaky-remote-shuffle-master-heartbeat.
Authored-by: afterincomparableyum
<[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/20
AI-Contributed/UT: 0/6
---
.../celeborn/plugin/flink/RemoteShuffleMasterSuiteJ.java | 6 ++----
.../org/apache/celeborn/client/ApplicationHeartbeater.scala | 8 +++++++-
.../main/scala/org/apache/celeborn/common/CelebornConf.scala | 12 ++++++++++++
3 files changed, 21 insertions(+), 5 deletions(-)
diff --git
a/client-flink/common/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterSuiteJ.java
b/client-flink/common/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterSuiteJ.java
index c0e73422bc..02e6e8039d 100644
---
a/client-flink/common/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterSuiteJ.java
+++
b/client-flink/common/src/test/java/org/apache/celeborn/plugin/flink/RemoteShuffleMasterSuiteJ.java
@@ -72,10 +72,8 @@ public class RemoteShuffleMasterSuiteJ {
int startPort = Utils$.MODULE$.selectRandomInt(1024, 65535);
configuration.setString("celeborn.master.port", String.valueOf(startPort));
configuration.setString("celeborn.master.endpoints", "localhost:" +
startPort);
- // Set a large app heartbeat interval to avoid the app heartbeater
resetting
- // LifecycleManager#shuffleCount/shuffleFallbackCounts via sumThenReset in
the
- // middle of a test, which makes the count assertions below flaky.
- configuration.setString("celeborn.client.application.heartbeatInterval",
"1h");
+ configuration.setString(
+ CelebornConf.TEST_CLIENT_APPLICATION_HEARTBEAT_ENABLED().key(),
"false");
remoteShuffleMaster = createShuffleMaster(configuration);
}
diff --git
a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
index e91e236b6c..84f4d07253 100644
---
a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
+++
b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala
@@ -56,6 +56,10 @@ class ApplicationHeartbeater(
private var appHeartbeat: ScheduledFuture[_] = _
def start(): Unit = {
+ if (!conf.testAppHeartbeatEnabled) {
+ logInfo(s"Application heartbeat of $appId is disabled for test.")
+ return
+ }
appHeartbeat = appHeartbeatHandlerThread.scheduleWithFixedDelay(
new Runnable {
override def run(): Unit = {
@@ -175,7 +179,9 @@ class ApplicationHeartbeater(
if (!stopped) {
// Stop appHeartbeat first
logInfo(s"Stop Application heartbeat $appId")
- appHeartbeat.cancel(true)
+ if (appHeartbeat != null) {
+ appHeartbeat.cancel(true)
+ }
ThreadUtils.shutdown(appHeartbeatHandlerThread)
if (applicationUnregisterEnabled) {
unregisterApplication()
diff --git
a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 22d1a066db..9b56cbc3a1 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -1521,6 +1521,7 @@ class CelebornConf(loadDefaults: Boolean) extends
Cloneable with Logging with Se
def testPushPrimaryDataTimeout: Boolean =
get(TEST_CLIENT_PUSH_PRIMARY_DATA_TIMEOUT)
def testPushReplicaDataTimeout: Boolean =
get(TEST_WORKER_PUSH_REPLICA_DATA_TIMEOUT)
def testRetryRevive: Boolean = get(TEST_CLIENT_RETRY_REVIVE)
+ def testAppHeartbeatEnabled: Boolean =
get(TEST_CLIENT_APPLICATION_HEARTBEAT_ENABLED)
def testAlternative: String = get(TEST_ALTERNATIVE.key, "celeborn")
def testProcessEndpointVerifierSeparate: Boolean =
get(TEST_RPC_ENDPOINT_VERIFIER_SEPARATE)
def clientFlinkMemoryPerResultPartition: Long =
get(CLIENT_MEMORY_PER_RESULT_PARTITION)
@@ -5275,6 +5276,17 @@ object CelebornConf extends Logging {
.booleanConf
.createWithDefault(false)
+ val TEST_CLIENT_APPLICATION_HEARTBEAT_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.test.client.application.heartbeat.enabled")
+ .internal
+ .categories("test", "client")
+ .doc("Whether LifecycleManager schedules the application heartbeat to
master. Only for " +
+ "tests without a master, whose assertions on shuffle metrics would
otherwise race with " +
+ "the heartbeat resetting them.")
+ .version("1.0.0")
+ .booleanConf
+ .createWithDefault(true)
+
val CLIENT_FETCH_TIMEOUT: ConfigEntry[Long] =
buildConf("celeborn.client.fetch.timeout")
.withAlternative("celeborn.fetch.timeout")
