SteNicholas commented on code in PR #3739:
URL: https://github.com/apache/celeborn/pull/3739#discussion_r3413163998
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -1636,4 +1643,16 @@ private[deploy] object Master extends Logging {
System.exit(-1)
}
}
+
+ private[master] def isClusterOverloaded(
+ conf: CelebornConf,
+ workersMap: java.util.Map[String, WorkerInfo],
+ availableWorkers: java.util.Set[WorkerInfo]): Boolean = {
+ if (!conf.clusterOverloadGcEnabled) return false
+ val totalCapacity = workersMap.values().asScala.map(_.totalSpace()).sum
+ if (totalCapacity <= 0) return false
Review Comment:
**Breaks on remote storage (S3/OSS/HDFS).** Remote `DiskInfo`s are created
with `actualUsableSpace = Long.MaxValue` and `totalSpace = 0`
(`StorageManager`'s `remoteDiskInfos`) and are reported to the master via
`allDisksSnapshot()` in every worker heartbeat. So `WorkerInfo.totalSpace()`
contributes `0` per remote disk while `totalActualUsableSpace()` contributes
`Long.MaxValue`.
- **Remote-only cluster:** every worker's `totalSpace()` is `0` →
`totalCapacity == 0` → this guard returns `false` permanently. The feature is a
silent no-op on exactly the deployments where stale-shuffle disk pressure
matters most.
- **Hybrid cluster:** `freeCapacity` (line 1654) sums `Long.MaxValue` per
available remote worker → Long overflow to a negative value → `usedFraction = 1
- neg/pos > 1` → reports overloaded every heartbeat (or, in the single-remote
no-overflow case, hugely negative → never).
Consider excluding remote / `Long.MaxValue` disks from this computation, or
scoping it to local-disk capacity only.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -1636,4 +1643,16 @@ private[deploy] object Master extends Logging {
System.exit(-1)
}
}
+
+ private[master] def isClusterOverloaded(
+ conf: CelebornConf,
+ workersMap: java.util.Map[String, WorkerInfo],
+ availableWorkers: java.util.Set[WorkerInfo]): Boolean = {
+ if (!conf.clusterOverloadGcEnabled) return false
+ val totalCapacity = workersMap.values().asScala.map(_.totalSpace()).sum
+ if (totalCapacity <= 0) return false
+ val freeCapacity =
availableWorkers.asScala.toList.map(_.totalActualUsableSpace()).sum
Review Comment:
**Numerator/denominator worker-set mismatch.** `freeCapacity` sums over
`availableWorkers` (a subset), but `totalCapacity` (line 1652) sums over all
`workersMap`. A worker excluded merely for **high workload** — which happens
precisely under load (`AbstractMetaManager.updateWorkerHeartbeatMeta` adds it
to `excludedWorkers` and drops it from `availableWorkers`) — keeps its full
`totalSpace` in the denominator but loses its free space from the numerator.
Example: 10 workers each 90% empty; load spikes and 9 report high workload →
`freeCapacity = 900GB`, `totalCapacity = 10TB` → `usedFraction = 0.91 ≥ 0.9` →
fleet-wide `System.gc()` on a 90%-empty cluster, adding STW pauses while it's
already busy. Same for shutdown/decommission/lost workers, none of which mean
'disk full'. Sum free and total over the **same** worker set.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -1636,4 +1643,16 @@ private[deploy] object Master extends Logging {
System.exit(-1)
}
}
+
+ private[master] def isClusterOverloaded(
Review Comment:
**Duplicates existing gauges + altitude.** This method's two sums are
verbatim copies of the `DEVICE_CELEBORN_TOTAL_CAPACITY` /
`DEVICE_CELEBORN_FREE_CAPACITY` gauge bodies (`Master.scala:297-303`). Extract
a shared `totalClusterCapacity()` / `freeClusterCapacity()` so the accounting
can't silently diverge from the published metrics.
On altitude: broadcasting `System.gc()` to every driver JVM is an indirect,
unmeasurable lever for stale-shuffle expiry (it frees nothing if references are
still held, and the STW cost lands on all drivers). Celeborn already models
'overload' via per-worker `highWorkload` +
`autoReleaseHighWorkLoadRatioThreshold`; a targeted shuffle-expiry path would
be more deterministic and reuse existing machinery.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -1636,4 +1643,16 @@ private[deploy] object Master extends Logging {
System.exit(-1)
}
}
+
+ private[master] def isClusterOverloaded(
+ conf: CelebornConf,
+ workersMap: java.util.Map[String, WorkerInfo],
+ availableWorkers: java.util.Set[WorkerInfo]): Boolean = {
+ if (!conf.clusterOverloadGcEnabled) return false
+ val totalCapacity = workersMap.values().asScala.map(_.totalSpace()).sum
+ if (totalCapacity <= 0) return false
+ val freeCapacity =
availableWorkers.asScala.toList.map(_.totalActualUsableSpace()).sum
+ val usedFraction = 1.0 - freeCapacity.toDouble / totalCapacity.toDouble
Review Comment:
**`totalSpace` and `totalActualUsableSpace` are on different scales.**
`totalSpace()` is the full filesystem size, but `totalActualUsableSpace()` is
capped at the worker dir's configured `capacity=` and is net of the disk
reserve (`StorageManager.updateDiskInfos`: `min(configuredUsableSpace - usage,
fsFree - reserve)`). When `capacity=` is set below the disk size, the disk is
shared with non-Celeborn data, or the reserve is large, `usedFraction` is
inflated.
E.g. `dir:capacity=200G` on a 2TB disk → `usedFraction ≈ 0.90` on a
brand-new **empty** cluster → reported overloaded forever. The default
`capacity` (1PB) avoids this, but `capacity=` is a supported, documented
setting. The doc ('Fraction of total cluster disk capacity used') doesn't match
this computation.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -1636,4 +1643,16 @@ private[deploy] object Master extends Logging {
System.exit(-1)
}
}
+
+ private[master] def isClusterOverloaded(
+ conf: CelebornConf,
+ workersMap: java.util.Map[String, WorkerInfo],
+ availableWorkers: java.util.Set[WorkerInfo]): Boolean = {
+ if (!conf.clusterOverloadGcEnabled) return false
+ val totalCapacity = workersMap.values().asScala.map(_.totalSpace()).sum
Review Comment:
**Unsynchronized read of mutable cluster state.** `workersMap` and
`availableWorkers` are mutated together under `synchronized(workersMap)` in
`AbstractMetaManager` (e.g. `removeWorkerMeta`: `workersMap.remove(...)` then
`availableWorkers.remove(...)`). This method reads both without that lock, so
it can observe a worker already gone from `workersMap` (smaller `totalCapacity`
here) but still present in `availableWorkers` (line 1654) → `freeCapacity >
totalCapacity` → negative `usedFraction` → a transient missed signal. The
existing `DEVICE_CELEBORN_*_CAPACITY` gauges share this property, but only feed
metrics, not a control decision.
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -170,6 +175,21 @@ class ApplicationHeartbeater(
}
}
+ private[client] def handleGcSignal(shouldTriggerGc: Boolean): Unit = {
+ if (!gcOnOverloadEnabled || !shouldTriggerGc) return
+ val now = System.currentTimeMillis()
+ if (now - lastGcTriggerTimeMs >= gcOnOverloadMinIntervalMs) {
Review Comment:
**Cooldown uses non-monotonic `System.currentTimeMillis()`.** A backward
clock step (NTP correction) after a GC makes `now - lastGcTriggerTimeMs`
negative on every subsequent heartbeat → always `< minInterval` → GC suppressed
for the whole jump duration, exactly when the cluster may be overloaded. A
forward step defeats the cooldown. Use `System.nanoTime()` for elapsed-time
comparisons.
##########
master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.service.deploy.master
+
+import java.util
+import java.util.concurrent.ConcurrentHashMap
+
+import scala.collection.JavaConverters._
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+import org.apache.celeborn.common.meta.{DiskInfo, WorkerInfo}
+
+class ClusterOverloadGcSuite extends CelebornFunSuite {
+
+ private def makeConf(enabled: Boolean, threshold: Double): CelebornConf = {
+ val conf = new CelebornConf()
+ conf.set(CelebornConf.MASTER_CLUSTER_OVERLOAD_GC_ENABLED, enabled)
+ conf.set(CelebornConf.MASTER_CLUSTER_OVERLOAD_GC_THRESHOLD, threshold)
+ conf
+ }
+
+ private def makeWorker(host: String, totalSpace: Long, usableSpace: Long):
WorkerInfo = {
+ val worker = new WorkerInfo(host, 10001, 10002, 10003, 10004)
+ val disk = new DiskInfo("/mnt/data", usableSpace, 0L, 0L, 0L)
+ disk.totalSpace = totalSpace
+ worker.updateThenGetDiskInfos(Map("/mnt/data" -> disk).asJava)
+ worker
+ }
+
+ private def workersMap(workers: WorkerInfo*): java.util.Map[String,
WorkerInfo] = {
+ val m = new util.HashMap[String, WorkerInfo]()
+ workers.foreach(w => m.put(w.toUniqueId, w))
+ m
+ }
+
+ private def availableWorkers(workers: WorkerInfo*):
java.util.Set[WorkerInfo] = {
+ val s = ConcurrentHashMap.newKeySet[WorkerInfo]()
+ workers.foreach(s.add)
+ s
+ }
+
+ test("returns false when feature is disabled") {
+ val conf = makeConf(enabled = false, threshold = 0.9)
+ val w = makeWorker("host1", totalSpace = 1000L, usableSpace = 10L)
+ assert(!Master.isClusterOverloaded(conf, workersMap(w),
availableWorkers(w)))
+ }
+
+ test("returns false when cluster is below the threshold") {
+ // 50% used — below 90% threshold
+ val conf = makeConf(enabled = true, threshold = 0.9)
+ val w = makeWorker("host1", totalSpace = 1000L, usableSpace = 500L)
+ assert(!Master.isClusterOverloaded(conf, workersMap(w),
availableWorkers(w)))
+ }
+
+ test("returns true when cluster is exactly at the threshold") {
+ // 90% used, 10% free — exactly at the 90% threshold
+ val conf = makeConf(enabled = true, threshold = 0.9)
+ val w = makeWorker("host1", totalSpace = 1000L, usableSpace = 100L)
+ assert(Master.isClusterOverloaded(conf, workersMap(w),
availableWorkers(w)))
+ }
+
+ test("returns true when cluster exceeds the threshold") {
+ // 95% used, 5% free — above 90% threshold
+ val conf = makeConf(enabled = true, threshold = 0.9)
+ val w = makeWorker("host1", totalSpace = 1000L, usableSpace = 50L)
+ assert(Master.isClusterOverloaded(conf, workersMap(w),
availableWorkers(w)))
+ }
+
+ test("returns false when totalCapacity is zero (no workers)") {
+ val conf = makeConf(enabled = true, threshold = 0.9)
+ assert(!Master.isClusterOverloaded(conf, workersMap(), availableWorkers()))
+ }
+
+ test("available workers can be a subset of all workers") {
+ // Two workers; only one is available for free-capacity accounting.
+ // total = 1000+1000 = 2000, free (available only) = 100 → 95% used →
overloaded.
+ val conf = makeConf(enabled = true, threshold = 0.9)
+ val w1 = makeWorker("host1", totalSpace = 1000L, usableSpace = 100L)
+ val w2 = makeWorker("host2", totalSpace = 1000L, usableSpace = 900L)
+ // Only w1 is available (e.g. w2 is excluded/shutdown)
+ assert(Master.isClusterOverloaded(conf, workersMap(w1, w2),
availableWorkers(w1)))
Review Comment:
**This test locks in the worker-set-mismatch bug.** It asserts
`isClusterOverloaded == true` for two healthy workers whose true utilization is
50% (`w2` is merely 'not available'). That result only holds because
`freeCapacity` (available subset) is divided by `totalCapacity` (all workers) —
the inconsistency flagged on the production method. When that's fixed to sum
both over the same worker set, this test will go red on a now-correct
implementation, so it effectively cements the wrong behavior.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2517,6 +2522,26 @@ object CelebornConf extends Logging {
.timeConf(TimeUnit.MILLISECONDS)
.createWithDefaultString("300s")
+ val MASTER_CLUSTER_OVERLOAD_GC_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.master.clusterOverload.gc.enabled")
+ .categories("master")
+ .version("0.7.0")
+ .doc("Whether to enable the master signaling clients to trigger GC when
the cluster " +
+ "storage is overloaded (disk usage exceeds the threshold).")
+ .booleanConf
+ .createWithDefault(false)
+
+ val MASTER_CLUSTER_OVERLOAD_GC_THRESHOLD: ConfigEntry[Double] =
+ buildConf("celeborn.master.clusterOverload.gc.threshold")
+ .categories("master")
+ .version("0.7.0")
+ .doc("Fraction of total cluster disk capacity used (0.0–1.0) above which
the master " +
+ "signals clients to trigger GC to release stale shuffle dependencies.
For example, " +
+ "0.9 means 90% of total capacity is in use.")
+ .doubleConf
+ .checkValue(v => v > 0.0 && v <= 1.0, "Must be between 0 (exclusive) and
1 (inclusive)")
Review Comment:
**Doc/validation mismatch.** The doc (line 2538) says the range is
'(0.0–1.0)', but `checkValue` rejects exactly `0.0` (`v > 0.0`). An operator
setting `0.0` (expecting 'always overloaded' per the doc) will fail config
validation. Align the doc to '(0.0 exclusive, 1.0 inclusive]'. Minor: the doc
uses a Unicode en-dash (`–`) rather than a hyphen.
##########
client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client
+
+import java.util.concurrent.ConcurrentHashMap
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+
+class ApplicationHeartbeaterSuite extends CelebornFunSuite {
+
+ private def makeHeartbeater(
+ gcEnabled: Boolean = true,
+ minIntervalMs: Long = 60000L): ApplicationHeartbeater = {
+ val conf = new CelebornConf()
+ conf.set(CelebornConf.CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED, gcEnabled)
+ conf.set(CelebornConf.CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL,
minIntervalMs)
+
+ val registeredShuffles = ConcurrentHashMap.newKeySet[Int]()
+ .asInstanceOf[ConcurrentHashMap.KeySetView[Int, java.lang.Boolean]]
+
+ new ApplicationHeartbeater(
+ "test-app",
+ conf,
+ null, // MasterClient not needed for handleGcSignal unit tests
+ () => (0L, 0L) -> (0L, 0L, Map.empty, Map.empty),
+ null, // WorkerStatusTracker not needed
+ registeredShuffles,
+ _ => ())
+ }
+
+ test("GC is not triggered when feature is disabled") {
+ val hb = makeHeartbeater(gcEnabled = false)
+ hb.handleGcSignal(shouldTriggerGc = true)
+ assert(hb.lastGcTriggerTimeMs == 0L)
+ }
+
+ test("GC is not triggered when signal is false") {
+ val hb = makeHeartbeater(gcEnabled = true)
+ hb.handleGcSignal(shouldTriggerGc = false)
+ assert(hb.lastGcTriggerTimeMs == 0L)
+ }
+
+ test("GC is triggered on first signal when enabled") {
+ val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 0L)
+ val before = System.currentTimeMillis()
+ hb.handleGcSignal(shouldTriggerGc = true)
+ assert(hb.lastGcTriggerTimeMs >= before)
+ }
+
+ test("GC is skipped when called again within the cooldown interval") {
+ val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 60000L)
+
+ hb.handleGcSignal(shouldTriggerGc = true)
+ val firstTrigger = hb.lastGcTriggerTimeMs
+ assert(firstTrigger > 0L)
+
+ // Second call immediately — well within 60s cooldown
+ hb.handleGcSignal(shouldTriggerGc = true)
+ assert(
+ hb.lastGcTriggerTimeMs == firstTrigger,
+ "lastGcTriggerTimeMs should not change on second call")
+ }
+
+ test("GC fires again after cooldown interval has elapsed") {
+ // Use 0ms cooldown so any elapsed time satisfies it.
+ val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 0L)
+
+ hb.handleGcSignal(shouldTriggerGc = true)
+ val firstTrigger = hb.lastGcTriggerTimeMs
+
+ // With 0ms interval the next call should always be allowed.
+ hb.handleGcSignal(shouldTriggerGc = true)
+ assert(hb.lastGcTriggerTimeMs >= firstTrigger)
Review Comment:
**Vacuous assertion.** `firstTrigger` is captured from
`lastGcTriggerTimeMs`, then after the second `handleGcSignal` the test asserts
`lastGcTriggerTimeMs >= firstTrigger`. If the re-trigger logic were broken and
GC never fired again, `lastGcTriggerTimeMs` stays `== firstTrigger`, so
`firstTrigger >= firstTrigger` is still true — the test passes whether or not
GC re-fires and cannot detect the regression it names. (With `minIntervalMs =
0L`, two calls can also land in the same millisecond, so even `>` would be
flaky.)
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala:
##########
@@ -1255,12 +1255,19 @@ private[celeborn] class Master(
new util.ArrayList[WorkerInfo](
(statusSystem.shutdownWorkers.asScala ++
statusSystem.decommissionWorkers.asScala).asJava),
new util.ArrayList(appRelatedShuffles),
- quotaManager.checkApplicationQuotaStatus(appId)))
+ quotaManager.checkApplicationQuotaStatus(appId),
+ shouldTriggerGcForApp()))
} else {
context.reply(OneWayMessageResponse)
}
}
+ private[master] def shouldTriggerGcForApp(): Boolean =
Review Comment:
**Recomputed per app heartbeat.** This yields a global, cluster-wide boolean
identical for all apps, yet it runs on every responding app heartbeat. Each
call sums over all workers twice, and `WorkerInfo.totalSpace()` /
`totalActualUsableSpace()` each take the worker's `this.synchronized` lock —
O(apps × workers) locked, contended work per heartbeat round for one boolean,
on the master RPC path. Compute it once per state-refresh tick and cache a
`@volatile Boolean`. (Note: `isClusterOverloaded` returns early when disabled,
so this cost only applies when the feature is on.)
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -170,6 +175,21 @@ class ApplicationHeartbeater(
}
}
+ private[client] def handleGcSignal(shouldTriggerGc: Boolean): Unit = {
+ if (!gcOnOverloadEnabled || !shouldTriggerGc) return
+ val now = System.currentTimeMillis()
+ if (now - lastGcTriggerTimeMs >= gcOnOverloadMinIntervalMs) {
+ logInfo(
+ "Cluster is overloaded; triggering System.gc() to release stale
shuffle dependencies.")
Review Comment:
**Silent no-op under `-XX:+DisableExplicitGC`.** With explicit GC disabled
(common on drivers), `System.gc()` (line 185) does nothing, yet this logs
'triggering System.gc()' and `lastGcTriggerTimeMs` is advanced as if it ran.
The feature is then silently ineffective with no diagnostic — stale shuffle is
never reclaimed and nothing in logs/metrics reveals why.
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -170,6 +175,21 @@ class ApplicationHeartbeater(
}
}
+ private[client] def handleGcSignal(shouldTriggerGc: Boolean): Unit = {
+ if (!gcOnOverloadEnabled || !shouldTriggerGc) return
+ val now = System.currentTimeMillis()
+ if (now - lastGcTriggerTimeMs >= gcOnOverloadMinIntervalMs) {
+ logInfo(
+ "Cluster is overloaded; triggering System.gc() to release stale
shuffle dependencies.")
+ lastGcTriggerTimeMs = now
+ System.gc()
Review Comment:
**Synchronous `System.gc()` on the single heartbeat thread.** The heartbeat
runnable is scheduled with `scheduleWithFixedDelay`, so a full STW collection
here (seconds–tens of seconds on a large driver heap) blocks the thread and
pushes out the next app heartbeat by the pause + `appHeartbeatIntervalMs`. If
the pause approaches the master's application-heartbeat timeout, the app can be
treated as lost — and this fires precisely under cluster overload. Consider
running the GC off-thread, or at least documenting the risk.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -4776,6 +4801,25 @@ object CelebornConf extends Logging {
.booleanConf
.createWithDefault(true)
+ val CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED: ConfigEntry[Boolean] =
+ buildConf("celeborn.client.clusterOverload.gc.enabled")
+ .categories("client")
+ .version("0.7.0")
+ .doc("When true, the client will trigger System.gc() upon receiving a GC
signal from " +
+ "the master indicating cluster storage is overloaded. Disable to
ignore the signal.")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
**Default asymmetry with the master flag.** This client flag defaults `true`
while `celeborn.master.clusterOverload.gc.enabled` defaults `false`. So
flipping only the master flag silently opts **every connected app** on default
client config into fleet-wide `System.gc()` STW pauses, with no action by app
owners. For an opt-in performance feature, consider defaulting this to `false`
(require both sides to opt in), or calling out prominently that the master flag
alone is fleet-wide-effective.
##########
client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala:
##########
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.celeborn.client
+
+import java.util.concurrent.ConcurrentHashMap
+
+import org.apache.celeborn.CelebornFunSuite
+import org.apache.celeborn.common.CelebornConf
+
+class ApplicationHeartbeaterSuite extends CelebornFunSuite {
+
+ private def makeHeartbeater(
+ gcEnabled: Boolean = true,
+ minIntervalMs: Long = 60000L): ApplicationHeartbeater = {
+ val conf = new CelebornConf()
+ conf.set(CelebornConf.CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED, gcEnabled)
+ conf.set(CelebornConf.CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL,
minIntervalMs)
+
+ val registeredShuffles = ConcurrentHashMap.newKeySet[Int]()
+ .asInstanceOf[ConcurrentHashMap.KeySetView[Int, java.lang.Boolean]]
+
+ new ApplicationHeartbeater(
Review Comment:
**Leaked executor per test.** Each `makeHeartbeater` constructs an
`ApplicationHeartbeater`, whose constructor eagerly starts a daemon scheduled
executor (`celeborn-client-lifecycle-manager-app-heartbeater`). No test calls
`stop()`, so each test in the suite leaks a live executor thread. They're
daemon threads so the JVM still exits, but under CI parallelism this is the
kind of teardown asymmetry that surfaces as 'unable to create new native
thread'. Add an `afterEach`/`stop()`, or share one instance.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
