This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch branch-0.7
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/branch-0.7 by this push:
new 5191a80f28 [CELEBORN-2487] Fix leaked LifecycleManager in spark-it
client suites
5191a80f28 is described below
commit 5191a80f2857eabc901c12f628dde6a06573d627
Author: yew1eb <[email protected]>
AuthorDate: Tue Oct 20 10:25:42 2026 +0800
[CELEBORN-2487] Fix leaked LifecycleManager in spark-it client suites
### What changes were proposed in this pull request?
Stop the LifecycleManager instances leaked by spark-it client suites:
- `LifecycleManagerSuite` "CELEBORN-1258" test: add
`lifecycleManager.stop()` at the end of the test.
- `ChangePartitionManagerUpdateWorkersSuite` "verify partition location
storage info" test: add `lifecycleManager.stop()` at the end of the test.
- `LifecycleManagerReserveSlotsSuite` (its only test): add
`lifecycleManager.stop()` at the end of the test.
- `ChangePartitionManagerUpdateWorkersSuite.afterEach()` now calls
`super.afterEach()` before shutting down the mini cluster, so the managers
created by `WithShuffleClientSuite`'s base tests (via `prepareService()`) are
stopped.
### Why are the changes needed?
Several spark-it suites create a `LifecycleManager` but never `stop()` it.
Each leaked manager's `ApplicationHeartbeater` keeps heartbeating the
already-stopped master for the rest of the test run, producing ~1200 ERROR
stack traces — about a third of the CI test log — which drowns out real
failures.
The `afterEach()` case is the largest source:
`ChangePartitionManagerUpdateWorkersSuite` overrides `afterEach()` to shut down
the mini cluster without calling `super`, so every base test's manager leaks
and errors every ~20s until JVM exit.
The cleanup follows the pattern already used by the other client suites
(e.g. `LifecycleManagerCommitFilesSuite`), which stop the manager at the end of
each test.
<img width="1287" height="615" alt="image"
src="https://github.com/user-attachments/assets/927603cd-2e2a-4062-8561-6050712dcccd";
/>
JIRA: [CELEBORN-2487](https://issues.apache.org/jira/browse/CELEBORN-2487)
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [ ] Yes
### How was this patch tested?
Ran the three affected suites with Spark 4.2:
```
build/mvn -Pspark-4.2,google-mirror -pl tests/spark-it test \
-DwildcardSuites=org.apache.celeborn.tests.client.LifecycleManagerSuite,org.apache.celeborn.tests.client.LifecycleManagerReserveSlotsSuite,org.apache.celeborn.tests.client.ChangePartitionManagerUpdateWorkersSuite
```
17 tests pass, and the `ApplicationHeartbeater: AskSync
HeartbeatFromApplication failed` / `MasterClient: Send rpc with failure` ERROR
stacks are gone (before the fix they were reproduced every ~20s until JVM exit).
Closes #3882 from yew1eb/CELEBORN-2487.
Authored-by: yew1eb <[email protected]>
Signed-off-by: zhengtao <[email protected]>
AI-Contributed/Feature: 0/0
AI-Contributed/UT: 0/10
(cherry picked from commit d7d6f7c5db6a60e786c4c50b137226fb501294c4)
Signed-off-by: zhengtao <[email protected]>
---
.../tests/client/ChangePartitionManagerUpdateWorkersSuite.scala | 6 ++++++
.../celeborn/tests/client/LifecycleManagerReserveSlotsSuite.scala | 2 ++
.../org/apache/celeborn/tests/client/LifecycleManagerSuite.scala | 2 ++
3 files changed, 10 insertions(+)
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
index 6e4b081230..912cac4d7a 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala
@@ -444,9 +444,15 @@ class ChangePartitionManagerUpdateWorkersSuite extends
WithShuffleClientSuite
shuffles.values().asScala.foreach(it =>
it.getPrimaryPartitions().asScala.foreach(loc =>
assert(loc.getStorageInfo.availableStorageTypes ==
StorageInfo.LOCAL_DISK_MASK)))
+
+ lifecycleManager.stop()
}
override def afterEach(): Unit = {
+ // stop the lifecycleManager/shuffleClient created by
WithShuffleClientSuite's
+ // base tests before shutting down the mini cluster, otherwise their
heartbeat
+ // threads keep erroring against the stopped master for the rest of the
run.
+ super.afterEach()
logInfo("test complete, stop celeborn mini cluster")
shutdownMiniCluster()
}
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerReserveSlotsSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerReserveSlotsSuite.scala
index 1626f2226b..684698839f 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerReserveSlotsSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerReserveSlotsSuite.scala
@@ -170,5 +170,7 @@ class LifecycleManagerReserveSlotsSuite extends AnyFunSuite
assert(partitionLocationMap2.get(partitions(0)).getEpoch > 0)
// epoch of partition(1) is 0 without split
assert(partitionLocationMap2.get(partitions(1)).getEpoch == 0)
+
+ lifecycleManager.stop()
}
}
diff --git
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala
index d34c79419b..b4453c83f9 100644
---
a/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala
+++
b/tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/LifecycleManagerSuite.scala
@@ -124,6 +124,8 @@ class LifecycleManagerSuite extends WithShuffleClientSuite
with MiniClusterFeatu
assert(appInfo.extraInfo.get("k1") == "v1")
assert(appInfo.registrationTime > 0 && appInfo.registrationTime <
System.currentTimeMillis())
}
+
+ lifecycleManager.stop()
}
override def afterAll(): Unit = {
