(celeborn) branch main updated: [CELEBORN-2347] Change get reducer file group cache to expireAfterAccess

[rexxiong]

This is an automated email from the ASF dual-hosted git repository.

RexXiong pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new a0f098f75 [CELEBORN-2347] Change get reducer file group cache to 
expireAfterAccess
a0f098f75 is described below

commit a0f098f75eff99ae4b914678c5eeacc089223889
Author: Sanskar Modi <sanskarmod...@gmail.com>
AuthorDate: Mon Jun 8 14:25:00 2026 +0800

    [CELEBORN-2347] Change get reducer file group cache to expireAfterAccess
    
    ### What changes were proposed in this pull request?
    
    Change get reducer file group cache to `expireAfterAccess`.
    
    ### Why are the changes needed?
    
    Currently the policy is expireAfterWrite which is not efficient, as it 
strictly clears the cache after the timeout, without considering that that 
entry was hot or not. `expireAfterAccess` will make sure to only clear if it 
was not actively not being accessed.
    
    ### Does this PR resolve a correctness bug?
    
    - [ ] Yes
    
    ### Does this PR introduce _any_ user-facing change?
    
    - [ ] Yes
    
    ### How was this patch tested?
    
    Existing UTs.
    
    Closes #3717 from s0nskar/cache_policy.
    
    Authored-by: Sanskar Modi <sanskarmod...@gmail.com>
    Signed-off-by: Shuang <lvshuang....@alibaba-inc.com>
---
 .../apache/celeborn/client/commit/ReducePartitionCommitHandler.scala   | 3 ++-
 common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala    | 2 +-
 docs/configuration/client.md                                           | 2 +-
 3 files changed, 4 insertions(+), 3 deletions(-)

diff --git 
a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
 
b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
index d88001c42..b3e7aa90a 100644
--- 
a/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
+++ 
b/client/src/main/scala/org/apache/celeborn/client/commit/ReducePartitionCommitHandler.scala
@@ -103,7 +103,7 @@ class ReducePartitionCommitHandler(
   // noinspection UnstableApiUsage
   private val getReducerFileGroupRpcCache: Cache[Int, ByteBuffer] = 
CacheBuilder.newBuilder()
     .concurrencyLevel(rpcCacheConcurrencyLevel)
-    .expireAfterWrite(rpcCacheExpireTime, TimeUnit.MILLISECONDS)
+    .expireAfterAccess(rpcCacheExpireTime, TimeUnit.MILLISECONDS)
     .maximumSize(rpcCacheSize)
     .build().asInstanceOf[Cache[Int, ByteBuffer]]
 
@@ -189,6 +189,7 @@ class ReducePartitionCommitHandler(
     shuffleIdLocks.remove(shuffleId)
     commitMetadataForReducer.remove(shuffleId)
     skewPartitionCompletenessValidator.remove(shuffleId)
+    getReducerFileGroupRpcCache.invalidate(shuffleId)
     super.removeExpiredShuffle(shuffleId)
   }
 
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 f81dacc56..668fbcf79 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -5620,7 +5620,7 @@ object CelebornConf extends Logging {
       .withAlternative("celeborn.rpc.cache.expireTime")
       .categories("client")
       .version("0.3.0")
-      .doc("The time before a cache item is removed.")
+      .doc("The idle time before a cache item is removed.")
       .timeConf(TimeUnit.MILLISECONDS)
       .createWithDefaultString("15s")
 
diff --git a/docs/configuration/client.md b/docs/configuration/client.md
index aec9fcc5f..5d15c6859 100644
--- a/docs/configuration/client.md
+++ b/docs/configuration/client.md
@@ -89,7 +89,7 @@ license: |
 | celeborn.client.reserveSlots.rackaware.enabled | false | false | Whether 
need to place different replicates on different racks when allocating slots. | 
0.3.1 | celeborn.client.reserveSlots.rackware.enabled | 
 | celeborn.client.reserveSlots.retryWait | 3s | false | Wait time before next 
retry if reserve slots failed. | 0.3.0 | celeborn.slots.reserve.retryWait | 
 | celeborn.client.rpc.cache.concurrencyLevel | 32 | false | The number of 
write locks to update rpc cache. | 0.3.0 | celeborn.rpc.cache.concurrencyLevel 
| 
-| celeborn.client.rpc.cache.expireTime | 15s | false | The time before a cache 
item is removed. | 0.3.0 | celeborn.rpc.cache.expireTime | 
+| celeborn.client.rpc.cache.expireTime | 15s | false | The idle time before a 
cache item is removed. | 0.3.0 | celeborn.rpc.cache.expireTime | 
 | celeborn.client.rpc.cache.size | 256 | false | The max cache items count for 
rpc cache. | 0.3.0 | celeborn.rpc.cache.size | 
 | celeborn.client.rpc.commitFiles.askTimeout | &lt;value of 
celeborn.rpc.askTimeout&gt; | false | Timeout for CommitHandler commit files. | 
0.4.1 |  | 
 | celeborn.client.rpc.getReducerFileGroup.askTimeout | &lt;value of 
celeborn.rpc.askTimeout&gt; | false | Timeout for ask operations during getting 
reducer file group information. During this process, there are 
`celeborn.client.requestCommitFiles.maxRetries` times for retry opportunities 
for committing files and 1 times for releasing slots request. User can 
customize this value according to your setting. | 0.2.0 |  |