[GLUTEN-12608][CORE] Make ShuffleManagerRouter cache tolerate the executor lifecycle

[LuciferYang]

What changes were proposed in this pull request?

ShuffleManagerRouter's inner Cache assumed each shuffleId is stored once, before any get or remove. That holds on the driver, where registerShuffle runs single-threaded from the DAGScheduler, but not on executors, where the cache is filled lazily and concurrently by task threads through getReader/getWriter. Two problems followed.

store used cache.compute with assert(m == null). When several tasks of the same new shuffleId first touch the cache at once on a multi-core executor, ConcurrentHashMap.compute serializes them and the later threads hit that assertion, so those tasks fail with an AssertionError. A retry succeeds once the cache is populated, but the failures still surface as flaky tasks. This change makes store an idempotent computeIfAbsent and takes the manager by name, so the lookup still runs only on a miss. The has() check in ensureShuffleManagerRegistered is then redundant, so it is removed.

remove used assert(manager != null). Spark broadcasts RemoveShuffle to every executor, so a router that never cached a shuffleId still gets unregisterShuffle and hits that assertion. remove now returns an Option, and unregisterShuffle reports that it removed nothing, matching SortShuffleManager, which returns a boolean instead of throwing.

Cache.get still asserts. Every path that reaches it runs on an executor that already registered the shuffleId (as writer via getWriter, or as reader via getReader, which registers before serving), so it is left unchanged.

How was this patch tested?

Added ShuffleManagerRouterCacheSuite with two tests:

• unregisterShuffle on a shuffleId this router never cached returns false instead of throwing.
• 4 threads over 200 iterations first-touching the same new shuffleId produce no errors.

Both fail on the current code (they hit the two assertions) and pass after the fix. The existing GlutenShuffleManagerSuite still passes.

Closes #12608

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Pull request overview

This pull request fixes two executor-side lifecycle assumptions in ShuffleManagerRouter’s per-executor cache that caused flaky task failures and noisy error logs: concurrent “first-touch” registration of a shuffleId and unregisterShuffle calls for shuffleIds that were never cached on a given executor.

Changes:

• Make Cache.store idempotent under concurrent first-touch by switching to computeIfAbsent and accepting the manager lookup by-name.
• Make Cache.remove miss-tolerant (Option) and update unregisterShuffle to return false on cache misses instead of asserting.
• Add a focused unit test suite reproducing both failure modes and validating the new behavior.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
gluten-core/src/main/scala/org/apache/spark/shuffle/ShuffleManagerRouter.scala	Makes the router cache tolerant of executor concurrency and “remove on uncached shuffleId” lifecycle by using computeIfAbsent and miss-safe removal.
gluten-core/src/test/scala/org/apache/spark/shuffle/ShuffleManagerRouterCacheSuite.scala	Adds regression tests for concurrent first-touch and unregisterShuffle cache-miss behavior.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[jackylee-ch]

👍