Copilot commented on code in PR #3739:
URL: https://github.com/apache/celeborn/pull/3739#discussion_r3593117838
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -47,6 +47,11 @@ class ApplicationHeartbeater(
private var stopped = false
private val reviseLostShuffles = conf.reviseLostShufflesEnabled
+ private val gcOnOverloadEnabled = conf.clientGcOnOverloadEnabled
+ private val gcOnOverloadMinIntervalMs = conf.clientGcOnOverloadMinIntervalMs
+ private val gcOnOverloadMinIntervalNs = gcOnOverloadMinIntervalMs * 1000000L
+ @volatile private var lastGcTriggerTimeNs = 0L
+
Review Comment:
`gcOnOverloadMinIntervalNs` is computed via `gcOnOverloadMinIntervalMs *
1000000L`, which can overflow for large intervals, and the class doesn't expose
any `lastGcTriggerTimeMs` value even though the new unit test expects it. Use
`TimeUnit.MILLISECONDS.toNanos(...)` for conversion and add a package-visible
`lastGcTriggerTimeMs` that tests can read.
##########
master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala:
##########
@@ -141,6 +153,23 @@ class QuotaManager(
checkQuotaSpace(CLUSTER_EXHAUSTED, consumption, getClusterStorageQuota)
}
+ // Calling the cluster overloaded (leads to gc triggers on app side to
relieve storage)
+ private def checkClusterOverloaded(consumption: ResourceConsumption):
Boolean = {
+ val overloadQuota = getClusterStorageQuota
+ val factor = getClusterOverloadLimitFactor
+ def scale(q: Long): Long = {
+ if (q <= 0L || q == Long.MaxValue) q
+ else math.max(1L, math.ceil(factor * q.toDouble).toLong)
+ }
+
+ val threshold = StorageQuota(
+ diskBytesWritten = scale(overloadQuota.diskBytesWritten),
+ diskFileCount = scale(overloadQuota.diskFileCount),
+ hdfsBytesWritten = scale(overloadQuota.hdfsBytesWritten),
+ hdfsFileCount = scale(overloadQuota.hdfsFileCount))
+ checkConsumptionExceeded(consumption, threshold)
+ }
Review Comment:
`checkClusterOverloaded` can report overload permanently if any cluster
quota dimension is configured as 0 (often used to mean "disabled"), because
`scale` keeps `q <= 0` as 0 and `checkConsumptionExceeded` uses `>=` without a
`> 0` guard. This would cause `shouldTriggerGc` to be true continuously even
when the cluster isn't actually constrained. Consider treating non-positive
thresholds as "ignored" in overload checks.
##########
client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala:
##########
@@ -170,6 +176,21 @@ class ApplicationHeartbeater(
}
}
+ private[client] def handleGcSignal(shouldTriggerGc: Boolean): Unit = {
+ if (!gcOnOverloadEnabled || !shouldTriggerGc) return
+ val nowNs = System.nanoTime()
+ if (nowNs - lastGcTriggerTimeNs >= gcOnOverloadMinIntervalNs) {
+ logInfo(
+ "Cluster is overloaded; triggering System.gc() to release stale
shuffle dependencies.")
+ lastGcTriggerTimeNs = nowNs
+ System.gc()
Review Comment:
When GC is triggered, only `lastGcTriggerTimeNs` is updated. The new unit
test asserts on `lastGcTriggerTimeMs`, so it should be set at the same time as
the successful GC trigger (using `System.currentTimeMillis()`), and remain
unchanged when the signal is skipped.
##########
docs/configuration/master.md:
##########
@@ -37,6 +37,7 @@ license: |
| celeborn.internal.port.enabled | false | false | Whether to create a
internal port on Masters/Workers for inter-Masters/Workers communication. This
is beneficial when SASL authentication is enforced for all interactions between
clients and Celeborn Services, but the services can exchange messages without
being subject to SASL authentication. | 0.5.0 | |
| celeborn.logConf.enabled | false | false | When `true`, log the CelebornConf
for debugging purposes. | 0.5.0 | |
| celeborn.master.allowWorkerHostPattern | <undefined> | false | Pattern
of worker host that allowed to register with the master. If not set, all
workers are allowed to register. | 0.6.0 | |
+| celeborn.master.clusterOverload.gc.enabled | false | false | Whether to
enable the master signaling clients to trigger GC when the cluster storage is
overloaded (disk usage exceeds the threshold). | 0.7.0 | |
Review Comment:
This description says overload is based on "disk usage", but the
implementation triggers overload when *any* configured cluster quota dimension
(disk bytes/files, HDFS bytes/files) crosses the overload threshold. Updating
the text will prevent operator confusion when overload is driven by non-disk
dimensions.
--
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]
