From fb8b66d466c6ce3aaa62ea0df251a52ef5a10e13 Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Mon, 15 Jun 2026 16:03:49 +0530 Subject: [PATCH 01/12] Add support to cleanup stale shuffle dependencies across all Spark apps when cluster under load --- .../client/ApplicationHeartbeater.scala | 19 ++++++++ common/src/main/proto/TransportMessages.proto | 1 + .../apache/celeborn/common/CelebornConf.scala | 44 +++++++++++++++++++ .../protocol/message/ControlMessages.scala | 10 +++-- .../common/util/PbSerDeUtilsTest.scala | 4 +- .../service/deploy/master/Master.scala | 13 +++++- 6 files changed, 86 insertions(+), 5 deletions(-) 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 e91e236b6c4..a8d12a949f5 100644 --- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala +++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala @@ -47,6 +47,10 @@ class ApplicationHeartbeater( private var stopped = false private val reviseLostShuffles = conf.reviseLostShufflesEnabled + private val gcOnOverloadEnabled = conf.clientGcOnOverloadEnabled + private val gcOnOverloadMinIntervalMs = conf.clientGcOnOverloadMinIntervalMs + @volatile private var lastGcTriggerTimeMs = 0L + // Use independent app heartbeat threads to avoid being blocked by other operations. private val appHeartbeatIntervalMs = conf.appHeartbeatIntervalMs private val applicationUnregisterEnabled = conf.applicationUnregisterEnabled @@ -91,6 +95,7 @@ class ApplicationHeartbeater( logDebug("Successfully send app heartbeat.") workerStatusTracker.handleHeartbeatResponse(response) checkQuotaExceeds(response.checkQuotaResponse) + handleGcSignal(response.shouldTriggerGc) // revise shuffle id if there are lost shuffles if (reviseLostShuffles) { val masterRecordedShuffleIds = response.registeredShuffles @@ -170,6 +175,20 @@ class ApplicationHeartbeater( } } + private 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() + } else { + logDebug( + s"Cluster overload GC signal received but skipped: last GC was triggered " + + s"${now - lastGcTriggerTimeMs}ms ago (min interval: ${gcOnOverloadMinIntervalMs}ms).") + } + } + def stop(): Unit = { this.synchronized { if (!stopped) { diff --git a/common/src/main/proto/TransportMessages.proto b/common/src/main/proto/TransportMessages.proto index a813a9e5015..223f981b7ca 100644 --- a/common/src/main/proto/TransportMessages.proto +++ b/common/src/main/proto/TransportMessages.proto @@ -510,6 +510,7 @@ message PbHeartbeatFromApplicationResponse { repeated PbWorkerInfo shuttingWorkers = 4; repeated int32 registeredShuffles = 5; PbCheckQuotaResponse checkQuotaResponse = 6; + bool shouldTriggerGc = 7; } message PbCheckQuota { 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 668fbcf79e4..cb069234165 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -936,6 +936,11 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se def userQuotaEnabled: Boolean = get(USER_QUOTA_ENABLED) def quotaInterruptShuffleEnabled: Boolean = get(QUOTA_INTERRUPT_SHUFFLE_ENABLED) + def clusterOverloadGcEnabled: Boolean = get(MASTER_CLUSTER_OVERLOAD_GC_ENABLED) + def clusterOverloadGcThreshold: Double = get(MASTER_CLUSTER_OVERLOAD_GC_THRESHOLD) + def clientGcOnOverloadEnabled: Boolean = get(CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED) + def clientGcOnOverloadMinIntervalMs: Long = get(CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL) + // ////////////////////////////////////////////////////// // Identity // // ////////////////////////////////////////////////////// @@ -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)") + .createWithDefault(0.9) + val DFS_EXPIRE_DIRS_TIMEOUT: ConfigEntry[Long] = buildConf("celeborn.master.dfs.expireDirs.timeout") .categories("master") @@ -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) + + val CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL: ConfigEntry[Long] = + buildConf("celeborn.client.clusterOverload.gc.minInterval") + .categories("client") + .version("0.7.0") + .doc("Minimum time that must elapse between consecutive GC triggers on the client side " + + "in response to master GC signals. This prevents excessive GC pressure when the " + + "cluster remains overloaded across multiple heartbeat intervals.") + .timeConf(TimeUnit.MILLISECONDS) + .createWithDefaultString("5m") + val CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED: ConfigEntry[Boolean] = buildConf("celeborn.client.excludePeerWorkerOnFailure.enabled") .categories("client") diff --git a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala index e12d4b697f7..31c4c1fb0d7 100644 --- a/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala +++ b/common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala @@ -397,7 +397,8 @@ object ControlMessages extends Logging { unknownWorkers: util.List[WorkerInfo], shuttingWorkers: util.List[WorkerInfo], registeredShuffles: util.List[Integer], - checkQuotaResponse: CheckQuotaResponse) extends Message + checkQuotaResponse: CheckQuotaResponse, + shouldTriggerGc: Boolean = false) extends Message case class CheckQuota(userIdentifier: UserIdentifier) extends Message @@ -890,7 +891,8 @@ object ControlMessages extends Logging { unknownWorkers, shuttingWorkers, registeredShuffles, - checkQuotaResponse) => + checkQuotaResponse, + shouldTriggerGc) => val pbCheckQuotaResponse = PbCheckQuotaResponse.newBuilder().setAvailable( checkQuotaResponse.isAvailable).setReason(checkQuotaResponse.reason) val payload = PbHeartbeatFromApplicationResponse.newBuilder() @@ -903,6 +905,7 @@ object ControlMessages extends Logging { shuttingWorkers.asScala.map(PbSerDeUtils.toPbWorkerInfo(_, true, true)).toList.asJava) .addAllRegisteredShuffles(registeredShuffles) .setCheckQuotaResponse(pbCheckQuotaResponse) + .setShouldTriggerGc(shouldTriggerGc) .build().toByteArray new TransportMessage(MessageType.HEARTBEAT_FROM_APPLICATION_RESPONSE, payload) @@ -1380,7 +1383,8 @@ object ControlMessages extends Logging { pbHeartbeatFromApplicationResponse.getShuttingWorkersList.asScala .map(PbSerDeUtils.fromPbWorkerInfo).toList.asJava, pbHeartbeatFromApplicationResponse.getRegisteredShufflesList, - checkQuotaResponse) + checkQuotaResponse, + pbHeartbeatFromApplicationResponse.getShouldTriggerGc) case CHECK_QUOTA_VALUE => val pbCheckAvailable = PbCheckQuota.parseFrom(message.getPayload) diff --git a/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala b/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala index 15c4df0fc71..a70413d1932 100644 --- a/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala +++ b/common/src/test/scala/org/apache/celeborn/common/util/PbSerDeUtilsTest.scala @@ -814,7 +814,8 @@ class PbSerDeUtilsTest extends CelebornFunSuite { mockWorkers("host1").toList.asJava, mockWorkers("host2").toList.asJava, Array(Integer.valueOf(1)).toList.asJava, - CheckQuotaResponse(isAvailable = false, "test_reason")) + CheckQuotaResponse(isAvailable = false, "test_reason"), + shouldTriggerGc = true) val toTransportHeartbeatFromApplicationResponse = ControlMessages.toTransportMessage(heartbeatFromApplicationResponse) val fromTransportHeartbeatFromApplicationResponse = @@ -822,6 +823,7 @@ class PbSerDeUtilsTest extends CelebornFunSuite { .asInstanceOf[HeartbeatFromApplicationResponse] assert(fromTransportHeartbeatFromApplicationResponse.equals(heartbeatFromApplicationResponse)) + assert(fromTransportHeartbeatFromApplicationResponse.shouldTriggerGc) } test("HeartbeatFromApplicationResponse backward compatibility without checkQuotaResponse") { diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index 6b25552060a..b56055564f1 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -1255,12 +1255,23 @@ 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 def shouldTriggerGcForApp(): Boolean = { + if (!conf.clusterOverloadGcEnabled) return false + val totalCapacity = statusSystem.workersMap.values().asScala.map(_.totalSpace()).sum + if (totalCapacity <= 0) return false + val freeCapacity = + statusSystem.availableWorkers.asScala.toList.map(_.totalActualUsableSpace()).sum + val usedFraction = 1.0 - freeCapacity.toDouble / totalCapacity.toDouble + usedFraction >= conf.clusterOverloadGcThreshold + } + private def handleRemoveWorkersUnavailableInfos( context: RpcCallContext, unavailableWorkers: util.List[WorkerInfo], From 0f148b7692c2b1a5a308a5264833cff6ce990d51 Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Mon, 15 Jun 2026 16:11:57 +0530 Subject: [PATCH 02/12] Lint fix --- .../org/apache/celeborn/client/ApplicationHeartbeater.scala | 3 ++- docs/configuration/client.md | 2 ++ docs/configuration/master.md | 2 ++ 3 files changed, 6 insertions(+), 1 deletion(-) 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 a8d12a949f5..8b0956015a8 100644 --- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala +++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala @@ -179,7 +179,8 @@ class ApplicationHeartbeater( if (!gcOnOverloadEnabled || !shouldTriggerGc) return val now = System.currentTimeMillis() if (now - lastGcTriggerTimeMs >= gcOnOverloadMinIntervalMs) { - logInfo("Cluster is overloaded; triggering System.gc() to release stale shuffle dependencies.") + logInfo( + "Cluster is overloaded; triggering System.gc() to release stale shuffle dependencies.") lastGcTriggerTimeMs = now System.gc() } else { diff --git a/docs/configuration/client.md b/docs/configuration/client.md index 5d15c6859fb..39ea80c035f 100644 --- a/docs/configuration/client.md +++ b/docs/configuration/client.md @@ -27,6 +27,8 @@ license: | | celeborn.client.application.uuidSuffix.enabled | false | false | Whether to add UUID suffix for application id for unique. When `true`, add UUID suffix for unique application id. Currently, this only applies to Spark and MR. | 0.6.0 | | | celeborn.client.chunk.prefetch.enabled | false | false | Whether to enable chunk prefetch when creating CelebornInputStream. | 0.5.1 | | | celeborn.client.closeIdleConnections | true | false | Whether client will close idle connections. | 0.3.0 | | +| celeborn.client.clusterOverload.gc.enabled | true | false | 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. | 0.7.0 | | +| celeborn.client.clusterOverload.gc.minInterval | 5m | false | Minimum time that must elapse between consecutive GC triggers on the client side in response to master GC signals. This prevents excessive GC pressure when the cluster remains overloaded across multiple heartbeat intervals. | 0.7.0 | | | celeborn.client.commitFiles.ignoreExcludedWorker | false | false | When true, LifecycleManager will skip workers which are in the excluded list. | 0.3.0 | | | celeborn.client.eagerlyCreateInputStream.threads | 32 | false | Threads count for streamCreatorPool in CelebornShuffleReader. | 0.3.1 | | | celeborn.client.excludePeerWorkerOnFailure.enabled | true | false | When true, Celeborn will exclude partition's peer worker on failure when push data to replica failed. | 0.3.0 | | diff --git a/docs/configuration/master.md b/docs/configuration/master.md index 1f1c900c946..5c1e1a9d5dd 100644 --- a/docs/configuration/master.md +++ b/docs/configuration/master.md @@ -37,6 +37,8 @@ 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 | | +| celeborn.master.clusterOverload.gc.threshold | 0.9 | false | 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. | 0.7.0 | | | celeborn.master.denyWorkerHostPattern | <undefined> | false | Pattern of worker host that denied to register with the master. If not set, no workers are denied to register. | 0.6.0 | | | celeborn.master.dfs.expireDirs.timeout | 1h | false | The timeout for an expired dirs to be deleted on dfs like HDFS, S3, OSS. | 0.6.0 | | | celeborn.master.estimatedPartitionSize.initialSize | 64mb | false | Initial partition size for estimation, it will change according to runtime stats. | 0.3.0 | celeborn.shuffle.initialEstimatedPartitionSize | From 15a7da980796a392de9d3e08cf16f78863f5248d Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Mon, 15 Jun 2026 16:31:29 +0530 Subject: [PATCH 03/12] Add ut --- .../client/ApplicationHeartbeater.scala | 4 +- .../client/ApplicationHeartbeaterSuite.scala | 112 +++++++++++++++++ .../service/deploy/master/Master.scala | 26 ++-- .../master/ClusterOverloadGcSuite.scala | 113 ++++++++++++++++++ 4 files changed, 244 insertions(+), 11 deletions(-) create mode 100644 client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala create mode 100644 master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala 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 8b0956015a8..b5f8dffe10f 100644 --- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala +++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala @@ -49,7 +49,7 @@ class ApplicationHeartbeater( private val gcOnOverloadEnabled = conf.clientGcOnOverloadEnabled private val gcOnOverloadMinIntervalMs = conf.clientGcOnOverloadMinIntervalMs - @volatile private var lastGcTriggerTimeMs = 0L + @volatile private[client] var lastGcTriggerTimeMs = 0L // Use independent app heartbeat threads to avoid being blocked by other operations. private val appHeartbeatIntervalMs = conf.appHeartbeatIntervalMs @@ -175,7 +175,7 @@ class ApplicationHeartbeater( } } - private def handleGcSignal(shouldTriggerGc: Boolean): Unit = { + private[client] def handleGcSignal(shouldTriggerGc: Boolean): Unit = { if (!gcOnOverloadEnabled || !shouldTriggerGc) return val now = System.currentTimeMillis() if (now - lastGcTriggerTimeMs >= gcOnOverloadMinIntervalMs) { diff --git a/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala b/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala new file mode 100644 index 00000000000..32877cf86c7 --- /dev/null +++ b/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) + } + + test("no GC when both signal is false and feature is disabled") { + val hb = makeHeartbeater(gcEnabled = false) + hb.handleGcSignal(shouldTriggerGc = false) + assert(hb.lastGcTriggerTimeMs == 0L) + } + + test("cooldown is measured from the last successful GC trigger") { + val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 60000L) + + // First trigger — sets lastGcTriggerTimeMs + hb.handleGcSignal(shouldTriggerGc = true) + val firstTrigger = hb.lastGcTriggerTimeMs + assert(firstTrigger > 0L) + + // Three more calls within cooldown — none should update lastGcTriggerTimeMs + hb.handleGcSignal(shouldTriggerGc = true) + hb.handleGcSignal(shouldTriggerGc = true) + hb.handleGcSignal(shouldTriggerGc = true) + assert(hb.lastGcTriggerTimeMs == firstTrigger) + } +} diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index b56055564f1..b5d28bf4a07 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -1262,15 +1262,11 @@ private[celeborn] class Master( } } - private def shouldTriggerGcForApp(): Boolean = { - if (!conf.clusterOverloadGcEnabled) return false - val totalCapacity = statusSystem.workersMap.values().asScala.map(_.totalSpace()).sum - if (totalCapacity <= 0) return false - val freeCapacity = - statusSystem.availableWorkers.asScala.toList.map(_.totalActualUsableSpace()).sum - val usedFraction = 1.0 - freeCapacity.toDouble / totalCapacity.toDouble - usedFraction >= conf.clusterOverloadGcThreshold - } + private[master] def shouldTriggerGcForApp(): Boolean = + Master.isClusterOverloaded( + conf, + statusSystem.workersMap, + statusSystem.availableWorkers) private def handleRemoveWorkersUnavailableInfos( context: RpcCallContext, @@ -1647,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 + usedFraction >= conf.clusterOverloadGcThreshold + } } diff --git a/master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala b/master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala new file mode 100644 index 00000000000..91502a868fd --- /dev/null +++ b/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))) + } + + test("not overloaded when enough free space across multiple workers") { + // Each worker has 50% free → aggregate 50% used, well below 90% + val conf = makeConf(enabled = true, threshold = 0.9) + val w1 = makeWorker("host1", totalSpace = 1000L, usableSpace = 500L) + val w2 = makeWorker("host2", totalSpace = 1000L, usableSpace = 500L) + assert(!Master.isClusterOverloaded(conf, workersMap(w1, w2), availableWorkers(w1, w2))) + } + + test("threshold of 1.0 is never exceeded when there is any free space") { + val conf = makeConf(enabled = true, threshold = 1.0) + val w = makeWorker("host1", totalSpace = 1000L, usableSpace = 1L) + assert(!Master.isClusterOverloaded(conf, workersMap(w), availableWorkers(w))) + } +} From ad8ee378a29530f1b930fcab6ac8d5fef086d98e Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Wed, 15 Jul 2026 15:33:17 +0530 Subject: [PATCH 04/12] Move overload to quota manager --- .../apache/celeborn/common/CelebornConf.scala | 11 ++++++++ .../celeborn/common/quota/StorageQuota.scala | 1 + .../service/deploy/master/Master.scala | 2 +- .../deploy/master/quota/QuotaManager.scala | 26 +++++++++++++++++++ .../common/service/config/DynamicConfig.java | 12 ++++++++- 5 files changed, 50 insertions(+), 2 deletions(-) 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 cb069234165..75a8c5432e6 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -6939,6 +6939,17 @@ object CelebornConf extends Logging { .checkValue(v => v >= 0.0 && v < 1.0, "Should be in [0.0, 1).") .createWithDefault(0.3) + val QUOTA_CLUSTER_OVERLOAD_FACTOR: ConfigEntry[Double] = + buildConf("celeborn.quota.overload.factor") + .categories("quota") + .dynamic + .doc("This config decides the quota * factor at which to consider the cluster 'overloaded." + + " When the cluster is overloaded, application heartbeat responses contain a signal to" + + " trigger a gc to clean up dangling shuffle dependencies") + .version("0.6.0") + .doubleConf + .createWithDefault(0.8) + val QUOTA_CLUSTER_DISK_BYTES_WRITTEN: ConfigEntry[Long] = buildConf("celeborn.quota.cluster.diskBytesWritten") .categories("quota") diff --git a/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala b/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala index 1a7a8e52abf..d99bcaf80b6 100644 --- a/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala +++ b/common/src/main/scala/org/apache/celeborn/common/quota/StorageQuota.scala @@ -37,4 +37,5 @@ case class StorageQuota( object StorageQuota { val DEFAULT_QUOTA = StorageQuota(Long.MaxValue, Long.MaxValue, Long.MaxValue, Long.MaxValue) + val CLUSTER_OVERLOAD_LIMIT_DEFAULT_MULTIPLIER: Double = 0.8 } diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index b5d28bf4a07..bdfd40c5bda 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -1256,7 +1256,7 @@ private[celeborn] class Master( (statusSystem.shutdownWorkers.asScala ++ statusSystem.decommissionWorkers.asScala).asJava), new util.ArrayList(appRelatedShuffles), quotaManager.checkApplicationQuotaStatus(appId), - shouldTriggerGcForApp())) + quotaManager.isClusterOverloaded)) } else { context.reply(OneWayMessageResponse) } diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala index 9b2f4d8818d..ada1173a091 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala @@ -49,6 +49,8 @@ class QuotaManager( val resourceConsumptionMetricsEnabled = celebornConf.masterResourceConsumptionMetricsEnabled @volatile var clusterQuotaStatus: QuotaStatus = new QuotaStatus() + @volatile + var clusterOverloaded: Boolean = false val appQuotaStatus: JMap[String, QuotaStatus] = JavaUtils.newConcurrentHashMap() val userResourceConsumptionCache: JMap[UserIdentifier, ResourceConsumption] = JavaUtils.newConcurrentHashMap() @@ -99,6 +101,10 @@ class QuotaManager( CheckQuotaResponse(!status.exceed, status.exceedReason) } + def isClusterOverloaded: Boolean = { + clusterOverloaded + } + def getUserStorageQuota(user: UserIdentifier): StorageQuota = { Option(configService) .map(_.getTenantUserConfigFromCache(user.tenantId, user.name).getUserStorageQuota) @@ -111,6 +117,12 @@ class QuotaManager( .getOrElse(StorageQuota.DEFAULT_QUOTA) } + def getClusterOverloadLimitFactor: Double = { + Option(configService) + .map(_.getSystemConfigFromCache.getClusterOverloadQuotaFactor) + .getOrElse(StorageQuota.CLUSTER_OVERLOAD_LIMIT_DEFAULT_MULTIPLIER) + } + def getClusterStorageQuota: StorageQuota = { Option(configService) .map(_.getSystemConfigFromCache.getClusterStorageQuota) @@ -141,6 +153,19 @@ 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 clusterOverloadFactor = getClusterOverloadLimitFactor + checkQuotaSpace("cluster overloaded", consumption, + new StorageQuota( + diskBytesWritten = (clusterOverloadFactor * overloadQuota.diskBytesWritten).toLong, + diskFileCount = (clusterOverloadFactor * overloadQuota.diskFileCount).toLong, + hdfsBytesWritten = (clusterOverloadFactor * overloadQuota.hdfsBytesWritten).toLong, + hdfsFileCount = (clusterOverloadFactor * overloadQuota.hdfsFileCount).toLong, + )).exceed + } + private def checkQuotaSpace( reason: String, consumption: ResourceConsumption, @@ -261,6 +286,7 @@ class QuotaManager( // Expire cluster level exceeded app except already expired app clusterQuotaStatus = checkClusterQuotaSpace(clusterResourceConsumption) + clusterOverloaded = checkClusterOverloaded(clusterResourceConsumption) if (interruptShuffleEnabled && clusterQuotaStatus.exceed) { checkClusterResourceConsumption( tenantResourceConsumptions, diff --git a/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java b/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java index d55145f331b..ec2b3b71592 100644 --- a/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java +++ b/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java @@ -169,6 +169,15 @@ public WorkerTagsMeta getWorkerTagsMeta() { ConfigType.STRING)); } + public Double getClusterOverloadQuotaFactor() { + return getValue( + CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR().key(), + CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR(), + Double.TYPE, + ConfigType.DOUBLE + ); + } + public StorageQuota getClusterStorageQuota() { return new StorageQuota( getValue( @@ -280,7 +289,8 @@ public enum ConfigType { BYTES, STRING, TIME_MS, - BOOLEAN + BOOLEAN, + DOUBLE } public static T convert(Class clazz, String value) { From e5af335e8cd43c1fa972086c5e054178d6e00c77 Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Wed, 15 Jul 2026 15:36:32 +0530 Subject: [PATCH 05/12] Remove dead code --- .../apache/celeborn/common/CelebornConf.scala | 12 ------------ .../service/deploy/master/Master.scala | 18 ------------------ 2 files changed, 30 deletions(-) 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 75a8c5432e6..558e16c3060 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -937,7 +937,6 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se def quotaInterruptShuffleEnabled: Boolean = get(QUOTA_INTERRUPT_SHUFFLE_ENABLED) def clusterOverloadGcEnabled: Boolean = get(MASTER_CLUSTER_OVERLOAD_GC_ENABLED) - def clusterOverloadGcThreshold: Double = get(MASTER_CLUSTER_OVERLOAD_GC_THRESHOLD) def clientGcOnOverloadEnabled: Boolean = get(CLIENT_GC_ON_CLUSTER_OVERLOAD_ENABLED) def clientGcOnOverloadMinIntervalMs: Long = get(CLIENT_GC_ON_CLUSTER_OVERLOAD_MIN_INTERVAL) @@ -2531,17 +2530,6 @@ object CelebornConf extends Logging { .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)") - .createWithDefault(0.9) - val DFS_EXPIRE_DIRS_TIMEOUT: ConfigEntry[Long] = buildConf("celeborn.master.dfs.expireDirs.timeout") .categories("master") diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index bdfd40c5bda..806767468c8 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -1262,12 +1262,6 @@ private[celeborn] class Master( } } - private[master] def shouldTriggerGcForApp(): Boolean = - Master.isClusterOverloaded( - conf, - statusSystem.workersMap, - statusSystem.availableWorkers) - private def handleRemoveWorkersUnavailableInfos( context: RpcCallContext, unavailableWorkers: util.List[WorkerInfo], @@ -1643,16 +1637,4 @@ 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 - usedFraction >= conf.clusterOverloadGcThreshold - } } From c7e9adebb7c28dc503404e6ae45b86ab4c20fd04 Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Wed, 15 Jul 2026 16:07:27 +0530 Subject: [PATCH 06/12] Remove test --- .../apache/celeborn/common/CelebornConf.scala | 2 +- .../service/deploy/master/Master.scala | 2 +- .../master/ClusterOverloadGcSuite.scala | 113 ------------------ 3 files changed, 2 insertions(+), 115 deletions(-) delete mode 100644 master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala 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 558e16c3060..5b8d3e0ca72 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -6934,7 +6934,7 @@ object CelebornConf extends Logging { .doc("This config decides the quota * factor at which to consider the cluster 'overloaded." + " When the cluster is overloaded, application heartbeat responses contain a signal to" + " trigger a gc to clean up dangling shuffle dependencies") - .version("0.6.0") + .version("0.7.0") .doubleConf .createWithDefault(0.8) diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala index 806767468c8..bf14e73b929 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala @@ -1256,7 +1256,7 @@ private[celeborn] class Master( (statusSystem.shutdownWorkers.asScala ++ statusSystem.decommissionWorkers.asScala).asJava), new util.ArrayList(appRelatedShuffles), quotaManager.checkApplicationQuotaStatus(appId), - quotaManager.isClusterOverloaded)) + if (conf.clusterOverloadGcEnabled) quotaManager.isClusterOverloaded else false)) } else { context.reply(OneWayMessageResponse) } diff --git a/master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala b/master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala deleted file mode 100644 index 91502a868fd..00000000000 --- a/master/src/test/scala/org/apache/celeborn/service/deploy/master/ClusterOverloadGcSuite.scala +++ /dev/null @@ -1,113 +0,0 @@ -/* - * 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))) - } - - test("not overloaded when enough free space across multiple workers") { - // Each worker has 50% free → aggregate 50% used, well below 90% - val conf = makeConf(enabled = true, threshold = 0.9) - val w1 = makeWorker("host1", totalSpace = 1000L, usableSpace = 500L) - val w2 = makeWorker("host2", totalSpace = 1000L, usableSpace = 500L) - assert(!Master.isClusterOverloaded(conf, workersMap(w1, w2), availableWorkers(w1, w2))) - } - - test("threshold of 1.0 is never exceeded when there is any free space") { - val conf = makeConf(enabled = true, threshold = 1.0) - val w = makeWorker("host1", totalSpace = 1000L, usableSpace = 1L) - assert(!Master.isClusterOverloaded(conf, workersMap(w), availableWorkers(w))) - } -} From 32dc65867bbdd8490ff9ed21ed41d965ee6fb345 Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Wed, 15 Jul 2026 16:27:04 +0530 Subject: [PATCH 07/12] lint fix --- docs/configuration/master.md | 1 - docs/configuration/quota.md | 1 + .../deploy/master/quota/QuotaManager.scala | 15 +- .../master/quota/QuotaManagerSuite.scala | 130 ++++++++++++++++++ .../common/service/config/DynamicConfig.java | 9 +- 5 files changed, 143 insertions(+), 13 deletions(-) diff --git a/docs/configuration/master.md b/docs/configuration/master.md index 5c1e1a9d5dd..bc57d37ae11 100644 --- a/docs/configuration/master.md +++ b/docs/configuration/master.md @@ -38,7 +38,6 @@ license: | | 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 | | -| celeborn.master.clusterOverload.gc.threshold | 0.9 | false | 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. | 0.7.0 | | | celeborn.master.denyWorkerHostPattern | <undefined> | false | Pattern of worker host that denied to register with the master. If not set, no workers are denied to register. | 0.6.0 | | | celeborn.master.dfs.expireDirs.timeout | 1h | false | The timeout for an expired dirs to be deleted on dfs like HDFS, S3, OSS. | 0.6.0 | | | celeborn.master.estimatedPartitionSize.initialSize | 64mb | false | Initial partition size for estimation, it will change according to runtime stats. | 0.3.0 | celeborn.shuffle.initialEstimatedPartitionSize | diff --git a/docs/configuration/quota.md b/docs/configuration/quota.md index 4174940e2a8..2d400929988 100644 --- a/docs/configuration/quota.md +++ b/docs/configuration/quota.md @@ -26,6 +26,7 @@ license: | | celeborn.quota.cluster.hdfsFileCount | 9223372036854775807 | true | Cluster level quota dynamic configuration for written hdfs file count. | 0.6.0 | | | celeborn.quota.enabled | true | false | When Master side sets to true, the master will enable to check the quota via QuotaManager. When Client side sets to true, LifecycleManager will request Master side to check whether the current user has enough quota before registration of shuffle. Fallback to the default shuffle service when Master side checks that there is no enough quota for current user. | 0.2.0 | | | celeborn.quota.interruptShuffle.enabled | false | false | Whether to enable interrupt shuffle when quota exceeds. | 0.6.0 | | +| celeborn.quota.overload.factor | 0.8 | true | This config decides the quota * factor at which to consider the cluster 'overloaded. When the cluster is overloaded, application heartbeat responses contain a signal to trigger a gc to clean up dangling shuffle dependencies | 0.7.0 | | | celeborn.quota.tenant.diskBytesWritten | 9223372036854775807b | true | Tenant level quota dynamic configuration for written disk bytes. | 0.5.0 | | | celeborn.quota.tenant.diskFileCount | 9223372036854775807 | true | Tenant level quota dynamic configuration for written disk file count. | 0.5.0 | | | celeborn.quota.tenant.enabled | true | false | Whether to enable tenant-level quota. | 0.6.0 | | diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala index ada1173a091..7281c5093d7 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala @@ -119,7 +119,7 @@ class QuotaManager( def getClusterOverloadLimitFactor: Double = { Option(configService) - .map(_.getSystemConfigFromCache.getClusterOverloadQuotaFactor) + .map(_.getSystemConfigFromCache.getClusterOverloadQuotaFactor.doubleValue()) .getOrElse(StorageQuota.CLUSTER_OVERLOAD_LIMIT_DEFAULT_MULTIPLIER) } @@ -157,13 +157,14 @@ class QuotaManager( private def checkClusterOverloaded(consumption: ResourceConsumption): Boolean = { val overloadQuota = getClusterStorageQuota val clusterOverloadFactor = getClusterOverloadLimitFactor - checkQuotaSpace("cluster overloaded", consumption, + checkQuotaSpace( + "cluster overloaded", + consumption, new StorageQuota( - diskBytesWritten = (clusterOverloadFactor * overloadQuota.diskBytesWritten).toLong, - diskFileCount = (clusterOverloadFactor * overloadQuota.diskFileCount).toLong, - hdfsBytesWritten = (clusterOverloadFactor * overloadQuota.hdfsBytesWritten).toLong, - hdfsFileCount = (clusterOverloadFactor * overloadQuota.hdfsFileCount).toLong, - )).exceed + diskBytesWritten = (clusterOverloadFactor * overloadQuota.diskBytesWritten).toLong, + diskFileCount = (clusterOverloadFactor * overloadQuota.diskFileCount).toLong, + hdfsBytesWritten = (clusterOverloadFactor * overloadQuota.hdfsBytesWritten).toLong, + hdfsFileCount = (clusterOverloadFactor * overloadQuota.hdfsFileCount).toLong)).exceed } private def checkQuotaSpace( diff --git a/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala b/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala index 425a3404b4f..ba4030b1f74 100644 --- a/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala +++ b/master/src/test/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManagerSuite.scala @@ -681,6 +681,136 @@ class QuotaManagerSuite extends CelebornFunSuite clearUserConsumption() } + test("test cluster overload - below threshold is not overloaded") { + val user = UserIdentifier("tenant_01", "Jerry") + // Cluster quota = 100GB disk, factor = 0.8 → overload threshold = 80GB + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // 70GB is below the 80GB threshold + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("70g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + + assertFalse(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - above threshold is overloaded") { + val user = UserIdentifier("tenant_01", "Jerry") + // Cluster quota = 100GB disk, factor = 0.8 → overload threshold = 80GB + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // 85GB is above the 80GB threshold + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - at exact threshold boundary is overloaded") { + val user = UserIdentifier("tenant_01", "Jerry") + // Cluster quota = 100GB disk, factor = 0.8 → overload threshold = 80GB + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // Consumption exactly at threshold (80GB): checkQuota uses value >= quota so this triggers overload + val thresholdBytes = (0.8 * Utils.byteStringAsBytes("100g")).toLong + addUserConsumption(user, ResourceConsumption(thresholdBytes, 0, 0, 0)) + quotaManager.updateResourceConsumption() + + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - status resets when consumption drops below threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // First: consumption above threshold + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertTrue(quotaManager.isClusterOverloaded) + + // Then: consumption drops below threshold + clearUserConsumption() + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("70g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertFalse(quotaManager.isClusterOverloaded) + + clearUserConsumption() + } + + test("test cluster overload - custom factor changes the effective threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + configService.refreshCache() + + // With factor=0.9, threshold = 90GB; 85GB should not trigger overload + conf.set("celeborn.quota.overload.factor", "0.9") + configService.refreshCache() + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertFalse(quotaManager.isClusterOverloaded) + clearUserConsumption() + + // Same 85GB would trigger overload with a lower factor=0.8 (threshold=80GB) + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + addUserConsumption(user, ResourceConsumption(Utils.byteStringAsBytes("85g"), 0, 0, 0)) + quotaManager.updateResourceConsumption() + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - triggered by hdfs bytes exceeding threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.cluster.hdfsBytesWritten", "50gb") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // Disk usage is fine (10GB < 80GB threshold) but hdfs usage (45GB) exceeds hdfs threshold (40GB) + addUserConsumption( + user, + ResourceConsumption(Utils.byteStringAsBytes("10g"), 0, Utils.byteStringAsBytes("45g"), 0)) + quotaManager.updateResourceConsumption() + + assertTrue(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + + test("test cluster overload - multiple dimensions, all below threshold") { + val user = UserIdentifier("tenant_01", "Jerry") + conf.set("celeborn.quota.cluster.diskBytesWritten", "100gb") + conf.set("celeborn.quota.cluster.diskFileCount", "10000") + conf.set("celeborn.quota.cluster.hdfsBytesWritten", "50gb") + conf.set("celeborn.quota.cluster.hdfsFileCount", "5000") + conf.set("celeborn.quota.overload.factor", "0.8") + configService.refreshCache() + + // All dimensions below their respective 80% thresholds + addUserConsumption( + user, + ResourceConsumption( + Utils.byteStringAsBytes("75g"), // disk: 75GB < 80GB threshold + 7000, // files: 7000 < 8000 threshold + Utils.byteStringAsBytes("35g"), // hdfs: 35GB < 40GB threshold + 3000 + ) + ) // hdfs files: 3000 < 4000 threshold + quotaManager.updateResourceConsumption() + + assertFalse(quotaManager.isClusterOverloaded) + clearUserConsumption() + } + def checkUserQuota(userIdentifier: UserIdentifier): CheckQuotaResponse = { quotaManager.checkUserQuotaStatus(userIdentifier) } diff --git a/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java b/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java index ec2b3b71592..ac0673996e6 100644 --- a/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java +++ b/service/src/main/java/org/apache/celeborn/server/common/service/config/DynamicConfig.java @@ -171,11 +171,10 @@ public WorkerTagsMeta getWorkerTagsMeta() { public Double getClusterOverloadQuotaFactor() { return getValue( - CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR().key(), - CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR(), - Double.TYPE, - ConfigType.DOUBLE - ); + CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR().key(), + CelebornConf.QUOTA_CLUSTER_OVERLOAD_FACTOR(), + Double.TYPE, + ConfigType.DOUBLE); } public StorageQuota getClusterStorageQuota() { From 3cee98c9113d3a9e255c77b17c7a21a4e24b2d84 Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Thu, 16 Jul 2026 11:22:51 +0530 Subject: [PATCH 08/12] Address comments --- .../apache/celeborn/common/CelebornConf.scala | 5 +++-- docs/configuration/quota.md | 2 +- .../deploy/master/quota/QuotaManager.scala | 21 +++++++++++-------- 3 files changed, 16 insertions(+), 12 deletions(-) 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 5b8d3e0ca72..1bb92bc52e2 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -6931,11 +6931,12 @@ object CelebornConf extends Logging { buildConf("celeborn.quota.overload.factor") .categories("quota") .dynamic - .doc("This config decides the quota * factor at which to consider the cluster 'overloaded." + + .doc("This config decides the quota * factor at which to consider the cluster 'overloaded'." + " When the cluster is overloaded, application heartbeat responses contain a signal to" + - " trigger a gc to clean up dangling shuffle dependencies") + " trigger a GC to clean up dangling shuffle dependencies") .version("0.7.0") .doubleConf + .checkValue(v => v >= 0.0 && v <= 1.0, "Should be in [0.0, 1.0].") .createWithDefault(0.8) val QUOTA_CLUSTER_DISK_BYTES_WRITTEN: ConfigEntry[Long] = diff --git a/docs/configuration/quota.md b/docs/configuration/quota.md index 2d400929988..3a2e68af42a 100644 --- a/docs/configuration/quota.md +++ b/docs/configuration/quota.md @@ -26,7 +26,7 @@ license: | | celeborn.quota.cluster.hdfsFileCount | 9223372036854775807 | true | Cluster level quota dynamic configuration for written hdfs file count. | 0.6.0 | | | celeborn.quota.enabled | true | false | When Master side sets to true, the master will enable to check the quota via QuotaManager. When Client side sets to true, LifecycleManager will request Master side to check whether the current user has enough quota before registration of shuffle. Fallback to the default shuffle service when Master side checks that there is no enough quota for current user. | 0.2.0 | | | celeborn.quota.interruptShuffle.enabled | false | false | Whether to enable interrupt shuffle when quota exceeds. | 0.6.0 | | -| celeborn.quota.overload.factor | 0.8 | true | This config decides the quota * factor at which to consider the cluster 'overloaded. When the cluster is overloaded, application heartbeat responses contain a signal to trigger a gc to clean up dangling shuffle dependencies | 0.7.0 | | +| celeborn.quota.overload.factor | 0.8 | true | This config decides the quota * factor at which to consider the cluster 'overloaded'. When the cluster is overloaded, application heartbeat responses contain a signal to trigger a GC to clean up dangling shuffle dependencies | 0.7.0 | | | celeborn.quota.tenant.diskBytesWritten | 9223372036854775807b | true | Tenant level quota dynamic configuration for written disk bytes. | 0.5.0 | | | celeborn.quota.tenant.diskFileCount | 9223372036854775807 | true | Tenant level quota dynamic configuration for written disk file count. | 0.5.0 | | | celeborn.quota.tenant.enabled | true | false | Whether to enable tenant-level quota. | 0.6.0 | | diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala index 7281c5093d7..512b3f0a649 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala @@ -156,15 +156,18 @@ class QuotaManager( // Calling the cluster overloaded (leads to gc triggers on app side to relieve storage) private def checkClusterOverloaded(consumption: ResourceConsumption): Boolean = { val overloadQuota = getClusterStorageQuota - val clusterOverloadFactor = getClusterOverloadLimitFactor - checkQuotaSpace( - "cluster overloaded", - consumption, - new StorageQuota( - diskBytesWritten = (clusterOverloadFactor * overloadQuota.diskBytesWritten).toLong, - diskFileCount = (clusterOverloadFactor * overloadQuota.diskFileCount).toLong, - hdfsBytesWritten = (clusterOverloadFactor * overloadQuota.hdfsBytesWritten).toLong, - hdfsFileCount = (clusterOverloadFactor * overloadQuota.hdfsFileCount).toLong)).exceed + 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) } private def checkQuotaSpace( From 522960457c32feec87961db55cf4c1b48e712e6a Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Thu, 16 Jul 2026 11:51:42 +0530 Subject: [PATCH 09/12] Comments --- .../celeborn/client/ApplicationHeartbeater.scala | 11 ++++++----- .../org/apache/celeborn/common/CelebornConf.scala | 3 ++- 2 files changed, 8 insertions(+), 6 deletions(-) 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 b5f8dffe10f..f9fa68adc5a 100644 --- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala +++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala @@ -49,7 +49,8 @@ class ApplicationHeartbeater( private val gcOnOverloadEnabled = conf.clientGcOnOverloadEnabled private val gcOnOverloadMinIntervalMs = conf.clientGcOnOverloadMinIntervalMs - @volatile private[client] var lastGcTriggerTimeMs = 0L + private val gcOnOverloadMinIntervalNs = gcOnOverloadMinIntervalMs * 1000000L + @volatile private var lastGcTriggerTimeNs = 0L // Use independent app heartbeat threads to avoid being blocked by other operations. private val appHeartbeatIntervalMs = conf.appHeartbeatIntervalMs @@ -177,16 +178,16 @@ class ApplicationHeartbeater( private[client] def handleGcSignal(shouldTriggerGc: Boolean): Unit = { if (!gcOnOverloadEnabled || !shouldTriggerGc) return - val now = System.currentTimeMillis() - if (now - lastGcTriggerTimeMs >= gcOnOverloadMinIntervalMs) { + val nowNs = System.nanoTime() + if (nowNs - lastGcTriggerTimeNs >= gcOnOverloadMinIntervalNs) { logInfo( "Cluster is overloaded; triggering System.gc() to release stale shuffle dependencies.") - lastGcTriggerTimeMs = now + lastGcTriggerTimeNs = nowNs System.gc() } else { logDebug( s"Cluster overload GC signal received but skipped: last GC was triggered " + - s"${now - lastGcTriggerTimeMs}ms ago (min interval: ${gcOnOverloadMinIntervalMs}ms).") + s"${(nowNs - lastGcTriggerTimeNs) / 1000000L}ms ago (min interval: ${gcOnOverloadMinIntervalMs}ms).") } } 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 1bb92bc52e2..0baf2226ccb 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -4806,6 +4806,7 @@ object CelebornConf extends Logging { "in response to master GC signals. This prevents excessive GC pressure when the " + "cluster remains overloaded across multiple heartbeat intervals.") .timeConf(TimeUnit.MILLISECONDS) + .checkValue(_ >= 0, "Should be >= 0.") .createWithDefaultString("5m") val CLIENT_EXCLUDE_PEER_WORKER_ON_FAILURE_ENABLED: ConfigEntry[Boolean] = @@ -6936,7 +6937,7 @@ object CelebornConf extends Logging { " trigger a GC to clean up dangling shuffle dependencies") .version("0.7.0") .doubleConf - .checkValue(v => v >= 0.0 && v <= 1.0, "Should be in [0.0, 1.0].") + .checkValue(v => v > 0.0 && v <= 1.0, "Should be in (0.0, 1.0].") .createWithDefault(0.8) val QUOTA_CLUSTER_DISK_BYTES_WRITTEN: ConfigEntry[Long] = From d8454d3549927c8c009856b64c9e1c7ebb77ad6f Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Thu, 16 Jul 2026 12:05:59 +0530 Subject: [PATCH 10/12] comments --- .../client/ApplicationHeartbeater.scala | 5 +++- .../client/ApplicationHeartbeaterSuite.scala | 26 +++++++++---------- .../apache/celeborn/common/CelebornConf.scala | 2 +- docs/configuration/master.md | 2 +- .../deploy/master/quota/QuotaManager.scala | 7 ++++- 5 files changed, 25 insertions(+), 17 deletions(-) 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 f9fa68adc5a..c9c3e9df0c9 100644 --- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala +++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala @@ -23,6 +23,7 @@ import java.util.function.Consumer import scala.collection.JavaConverters._ +import com.google.common.annotations.VisibleForTesting import org.apache.commons.lang3.StringUtils import org.apache.celeborn.common.CelebornConf @@ -50,7 +51,9 @@ class ApplicationHeartbeater( private val gcOnOverloadEnabled = conf.clientGcOnOverloadEnabled private val gcOnOverloadMinIntervalMs = conf.clientGcOnOverloadMinIntervalMs private val gcOnOverloadMinIntervalNs = gcOnOverloadMinIntervalMs * 1000000L - @volatile private var lastGcTriggerTimeNs = 0L + + @VisibleForTesting + @volatile private[client] var lastGcTriggerTimeNs = 0L // Use independent app heartbeat threads to avoid being blocked by other operations. private val appHeartbeatIntervalMs = conf.appHeartbeatIntervalMs diff --git a/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala b/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala index 32877cf86c7..dda56dad07a 100644 --- a/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala +++ b/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala @@ -47,33 +47,33 @@ class ApplicationHeartbeaterSuite extends CelebornFunSuite { test("GC is not triggered when feature is disabled") { val hb = makeHeartbeater(gcEnabled = false) hb.handleGcSignal(shouldTriggerGc = true) - assert(hb.lastGcTriggerTimeMs == 0L) + assert(hb.lastGcTriggerTimeNs == 0L) } test("GC is not triggered when signal is false") { val hb = makeHeartbeater(gcEnabled = true) hb.handleGcSignal(shouldTriggerGc = false) - assert(hb.lastGcTriggerTimeMs == 0L) + assert(hb.lastGcTriggerTimeNs == 0L) } test("GC is triggered on first signal when enabled") { val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 0L) - val before = System.currentTimeMillis() + val before = System.nanoTime() hb.handleGcSignal(shouldTriggerGc = true) - assert(hb.lastGcTriggerTimeMs >= before) + assert(hb.lastGcTriggerTimeNs >= 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 + val firstTrigger = hb.lastGcTriggerTimeNs assert(firstTrigger > 0L) // Second call immediately — well within 60s cooldown hb.handleGcSignal(shouldTriggerGc = true) assert( - hb.lastGcTriggerTimeMs == firstTrigger, + hb.lastGcTriggerTimeNs == firstTrigger, "lastGcTriggerTimeMs should not change on second call") } @@ -82,31 +82,31 @@ class ApplicationHeartbeaterSuite extends CelebornFunSuite { val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 0L) hb.handleGcSignal(shouldTriggerGc = true) - val firstTrigger = hb.lastGcTriggerTimeMs + val firstTrigger = hb.lastGcTriggerTimeNs // With 0ms interval the next call should always be allowed. hb.handleGcSignal(shouldTriggerGc = true) - assert(hb.lastGcTriggerTimeMs >= firstTrigger) + assert(hb.lastGcTriggerTimeNs >= firstTrigger) } test("no GC when both signal is false and feature is disabled") { val hb = makeHeartbeater(gcEnabled = false) hb.handleGcSignal(shouldTriggerGc = false) - assert(hb.lastGcTriggerTimeMs == 0L) + assert(hb.lastGcTriggerTimeNs == 0L) } test("cooldown is measured from the last successful GC trigger") { val hb = makeHeartbeater(gcEnabled = true, minIntervalMs = 60000L) - // First trigger — sets lastGcTriggerTimeMs + // First trigger — sets lastGcTriggerTimeNs hb.handleGcSignal(shouldTriggerGc = true) - val firstTrigger = hb.lastGcTriggerTimeMs + val firstTrigger = hb.lastGcTriggerTimeNs assert(firstTrigger > 0L) - // Three more calls within cooldown — none should update lastGcTriggerTimeMs + // Three more calls within cooldown — none should update lastGcTriggerTimeNs hb.handleGcSignal(shouldTriggerGc = true) hb.handleGcSignal(shouldTriggerGc = true) hb.handleGcSignal(shouldTriggerGc = true) - assert(hb.lastGcTriggerTimeMs == firstTrigger) + assert(hb.lastGcTriggerTimeNs == firstTrigger) } } 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 0baf2226ccb..b525f8c2e14 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -2526,7 +2526,7 @@ object CelebornConf extends Logging { .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).") + "storage is overloaded (any cluster quota dimension exceeds the overload threshold).") .booleanConf .createWithDefault(false) diff --git a/docs/configuration/master.md b/docs/configuration/master.md index bc57d37ae11..302fb32617f 100644 --- a/docs/configuration/master.md +++ b/docs/configuration/master.md @@ -37,7 +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 | | +| celeborn.master.clusterOverload.gc.enabled | false | false | Whether to enable the master signaling clients to trigger GC when the cluster storage is overloaded (any cluster quota dimension exceeds the overload threshold). | 0.7.0 | | | celeborn.master.denyWorkerHostPattern | <undefined> | false | Pattern of worker host that denied to register with the master. If not set, no workers are denied to register. | 0.6.0 | | | celeborn.master.dfs.expireDirs.timeout | 1h | false | The timeout for an expired dirs to be deleted on dfs like HDFS, S3, OSS. | 0.6.0 | | | celeborn.master.estimatedPartitionSize.initialSize | 64mb | false | Initial partition size for estimation, it will change according to runtime stats. | 0.3.0 | celeborn.shuffle.initialEstimatedPartitionSize | diff --git a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala index 512b3f0a649..b03ffcb0377 100644 --- a/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala +++ b/master/src/main/scala/org/apache/celeborn/service/deploy/master/quota/QuotaManager.scala @@ -167,7 +167,12 @@ class QuotaManager( diskFileCount = scale(overloadQuota.diskFileCount), hdfsBytesWritten = scale(overloadQuota.hdfsBytesWritten), hdfsFileCount = scale(overloadQuota.hdfsFileCount)) - checkConsumptionExceeded(consumption, threshold) + def exceeded(used: Long, limit: Long): Boolean = limit > 0L && used >= limit + + exceeded(consumption.diskBytesWritten, threshold.diskBytesWritten) || + exceeded(consumption.diskFileCount, threshold.diskFileCount) || + exceeded(consumption.hdfsBytesWritten, threshold.hdfsBytesWritten) || + exceeded(consumption.hdfsFileCount, threshold.hdfsFileCount) } private def checkQuotaSpace( From 14173ade892a53f6420a973ad865fda498fec16f Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Thu, 16 Jul 2026 12:14:13 +0530 Subject: [PATCH 11/12] comments --- .../org/apache/celeborn/client/ApplicationHeartbeater.scala | 2 +- .../apache/celeborn/client/ApplicationHeartbeaterSuite.scala | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 c9c3e9df0c9..3ac3920bc81 100644 --- a/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala +++ b/client/src/main/scala/org/apache/celeborn/client/ApplicationHeartbeater.scala @@ -50,7 +50,7 @@ class ApplicationHeartbeater( private val gcOnOverloadEnabled = conf.clientGcOnOverloadEnabled private val gcOnOverloadMinIntervalMs = conf.clientGcOnOverloadMinIntervalMs - private val gcOnOverloadMinIntervalNs = gcOnOverloadMinIntervalMs * 1000000L + private val gcOnOverloadMinIntervalNs = TimeUnit.MILLISECONDS.toNanos(gcOnOverloadMinIntervalMs) @VisibleForTesting @volatile private[client] var lastGcTriggerTimeNs = 0L diff --git a/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala b/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala index dda56dad07a..69a1e47dd4d 100644 --- a/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala +++ b/client/src/test/scala/org/apache/celeborn/client/ApplicationHeartbeaterSuite.scala @@ -74,7 +74,7 @@ class ApplicationHeartbeaterSuite extends CelebornFunSuite { hb.handleGcSignal(shouldTriggerGc = true) assert( hb.lastGcTriggerTimeNs == firstTrigger, - "lastGcTriggerTimeMs should not change on second call") + "lastGcTriggerTimeNs should not change on second call") } test("GC fires again after cooldown interval has elapsed") { From 0542bcb3b9b47338401fd1d0c9c676d8c30a3cda Mon Sep 17 00:00:00 2001 From: Saurabh Dubey Date: Thu, 16 Jul 2026 12:51:45 +0530 Subject: [PATCH 12/12] Comments --- .../main/scala/org/apache/celeborn/common/CelebornConf.scala | 4 ++-- docs/configuration/quota.md | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) 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 b525f8c2e14..fe04f82dfb8 100644 --- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala +++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala @@ -6933,8 +6933,8 @@ object CelebornConf extends Logging { .categories("quota") .dynamic .doc("This config decides the quota * factor at which to consider the cluster 'overloaded'." + - " When the cluster is overloaded, application heartbeat responses contain a signal to" + - " trigger a GC to clean up dangling shuffle dependencies") + " When the cluster is overloaded and `celeborn.master.clusterOverload.gc.enabled` is true," + + " application heartbeat responses contain a signal to trigger a GC to clean up dangling shuffle dependencies") .version("0.7.0") .doubleConf .checkValue(v => v > 0.0 && v <= 1.0, "Should be in (0.0, 1.0].") diff --git a/docs/configuration/quota.md b/docs/configuration/quota.md index 3a2e68af42a..111492e6ec5 100644 --- a/docs/configuration/quota.md +++ b/docs/configuration/quota.md @@ -26,7 +26,7 @@ license: | | celeborn.quota.cluster.hdfsFileCount | 9223372036854775807 | true | Cluster level quota dynamic configuration for written hdfs file count. | 0.6.0 | | | celeborn.quota.enabled | true | false | When Master side sets to true, the master will enable to check the quota via QuotaManager. When Client side sets to true, LifecycleManager will request Master side to check whether the current user has enough quota before registration of shuffle. Fallback to the default shuffle service when Master side checks that there is no enough quota for current user. | 0.2.0 | | | celeborn.quota.interruptShuffle.enabled | false | false | Whether to enable interrupt shuffle when quota exceeds. | 0.6.0 | | -| celeborn.quota.overload.factor | 0.8 | true | This config decides the quota * factor at which to consider the cluster 'overloaded'. When the cluster is overloaded, application heartbeat responses contain a signal to trigger a GC to clean up dangling shuffle dependencies | 0.7.0 | | +| celeborn.quota.overload.factor | 0.8 | true | This config decides the quota * factor at which to consider the cluster 'overloaded'. When the cluster is overloaded and `celeborn.master.clusterOverload.gc.enabled` is true, application heartbeat responses contain a signal to trigger a GC to clean up dangling shuffle dependencies | 0.7.0 | | | celeborn.quota.tenant.diskBytesWritten | 9223372036854775807b | true | Tenant level quota dynamic configuration for written disk bytes. | 0.5.0 | | | celeborn.quota.tenant.diskFileCount | 9223372036854775807 | true | Tenant level quota dynamic configuration for written disk file count. | 0.5.0 | | | celeborn.quota.tenant.enabled | true | false | Whether to enable tenant-level quota. | 0.6.0 | |