[CELEBORN-2398] Allow ChangePartitionManager to refresh candidate workers without allocating slots

[Kalvin2077]

What changes were proposed in this pull request?

This PR introduces a Protobuf-based RequestWorkers/RequestWorkersResponse RPC for read-only worker discovery.

On the Master side, the new handler selects currently available workers after applying client exclusions and tag filters. It limits the response using the smaller of celeborn.client.slot.assign.maxWorkers and celeborn.master.splitSlot.assign.maxWorkers, while preserving the minimum worker count required for replication. When authentication is enabled, application metadata is pushed to the selected workers.

On the client side, LifecycleManager periodically requests workers from the Master, creates endpoints for newly selected workers, records connection failures, and maintains an endpoint-ready worker pool. ChangePartitionManager uses that pool for change-partition requests and falls back to the shuffle's existing worker snapshots when no refreshed candidates are available.

The PR also makes the following configuration changes:

• removes celeborn.client.shuffle.dynamicResourceFactor;
• adds celeborn.client.shuffle.dynamicResource.updateTime, defaulting to 30s;
• adds celeborn.master.splitSlot.assign.maxWorkers, defaulting to 500;
• documents the breaking client configuration change in the migration guide.

Why are the changes needed?

The existing factor-based logic refreshes candidates only after enough workers from the shuffle's original allocation become unavailable. It therefore cannot use newly added workers during normal workers scale-out while the original workers remain healthy.

Requesting slots merely to discover workers also mixes worker discovery with slot allocation and mutates Master shuffle state. A dedicated read-only RPC allows ChangePartitionManager to use the current cluster membership without those side effects, while rate limiting and worker-count caps prevent excessive RPC connections and oversized responses.

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

• Yes

Users of celeborn.client.shuffle.dynamicResourceFactor must migrate to celeborn.client.shuffle.dynamicResource.updateTime.

How was this patch tested?

• Added unit tests for Worker selection, storage eligibility, dynamic Worker merging, deduplication, and excluded-Worker filtering.
• Built with JDK 8 and deployed to a four-Worker Celeborn cluster.
• Ran an E2E shuffle that started with one Worker, added three Workers, and paused the original Worker. The client detected the failure, revived partitions on new Workers, and successfully validated 8 million rows with fallback disabled.
• Restored all Workers and reran the standard Spark/YARN smoke test successfully.

[zaynt4606]

Minor: storageTypes is likely a Set, so .head returns an arbitrary element. When multiple storage types are configured (e.g. HDD,MEMORY), the type sent to Master depends on Set iteration order. If .head returns a non-disk type, Master skips the haveDisk filter and may return disk-less workers. The impact is low — requestSlots does the real filtering — but with many application splits the extra unnecessary endpoints could add up. Consider picking a disk type preferentially, e.g. storageTypes.find(t => t == HDD || t == SSD).getOrElse(...).

[zaynt4606]

Minor: scala.util.Random is thread-safe but uses internal synchronization, which can cause contention under concurrent Master requests. Since handleRequestWorkers is throttled to 30s per client the practical impact is low, but if the number of active applications grows, consider using ThreadLocalRandom.current().nextInt(eligibleWorkers.size) for better concurrency. The rest of the file already uses Random.nextInt (lines 972, 1051) so this is consistent with existing code — not blocking.

[zaynt4606]

LGTM. The refactor cleanly decouples worker discovery from slot allocation, replacing the factor-based trigger with an on-demand, rate-limited read-only RPC. Concurrency handling in refreshEndpointReadyWorkersFromMaster is correct, tests are thorough, and the breaking config change is documented in the migration guide.

Two minor suggestions left as inline comments (non-blocking): storageTypes.head non-determinism and Random.nextInt concurrency. Neither warrants changes before merge.

[codecov]

Codecov Report

❌ Patch coverage is 31.06061% with 91 lines in your changes missing coverage. Please review.
✅ Project coverage is 58.58%. Comparing base (07dde50) to head (78cac4b).
⚠️ Report is 11 commits behind head on main.

Files with missing lines	Patch %	Lines
.../org/apache/celeborn/client/LifecycleManager.scala	0.00%	87 Missing ⚠️
...pache/celeborn/client/ChangePartitionManager.scala	71.43%	2 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #3775      +/-   ##
============================================
+ Coverage     58.41%   58.58%   +0.18%     
- Complexity      229      242      +13     
============================================
  Files           398      399       +1     
  Lines         27993    28128     +135     
  Branches       2734     2746      +12     
============================================
+ Hits          16349    16476     +127     
- Misses        10446    10448       +2     
- Partials       1198     1204       +6     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Copilot]

Pull request overview

This PR adds a new read-only Master RPC (RequestWorkers / RequestWorkersResponse) to support dynamic worker discovery without allocating slots, and updates client-side logic so ChangePartitionManager can refresh worker candidates via an endpoint-ready worker pool (throttled by a new update-time config). It also removes the previous factor-based refresh trigger and documents the migration.

Changes:

• Introduces Protobuf transport + ControlMessages plumbing for REQUEST_WORKERS and implements Master.handleRequestWorkers with filtering, limits, and optional auth metadata push.
• Updates client dynamic-resource refresh to be time-throttled (celeborn.client.shuffle.dynamicResource.updateTime) and to maintain an endpoint-ready worker set used by ChangePartitionManager.
• Updates tests and documentation to remove dynamicResourceFactor and validate the new request/refresh behavior.

Reviewed changes

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

Show a summary per file

File	Description
tests/spark-it/src/test/scala/org/apache/celeborn/tests/spark/RetryReviveTest.scala	Migrates IT config from removed factor setting to the new update-time setting.
tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala	Updates IT coverage for dynamic refresh, throttling, and candidate merging semantics.
master/src/test/scala/org/apache/celeborn/service/deploy/master/MasterSuite.scala	Adds unit tests for handleRequestWorkers selection/limits/eligibility behavior.
master/src/main/scala/org/apache/celeborn/service/deploy/master/Master.scala	Implements RequestWorkers handling and refactors auth metadata push helper.
docs/migration.md	Documents the breaking removal of dynamicResourceFactor and migration path.
docs/configuration/master.md	Documents new celeborn.master.splitSlot.assign.maxWorkers setting.
docs/configuration/client.md	Documents new celeborn.client.shuffle.dynamicResource.updateTime setting and updates dynamic refresh description.
common/src/test/scala/org/apache/celeborn/common/util/UtilsSuite.scala	Adds serialization tests for the new Protobuf messages via TransportMessage.
common/src/test/scala/org/apache/celeborn/common/CelebornConfSuite.scala	Adds validation tests for new/updated config entries.
common/src/main/scala/org/apache/celeborn/common/protocol/message/ControlMessages.scala	Adds TransportMessage encode/decode support for REQUEST_WORKERS and response.
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala	Adds new config entries and removes the factor-based config accessor/entry.
common/src/main/proto/TransportMessages.proto	Adds PbRequestWorkers and PbRequestWorkersResponse messages and message types.
client/src/test/scala/org/apache/celeborn/client/WorkerStatusTrackerSuite.scala	Adds unit test coverage for endpoint-ready worker pool maintenance.
client/src/test/scala/org/apache/celeborn/client/ChangePartitionManagerSuite.scala	Adds unit tests for candidate collection merging/deduplication and exclusion filtering.
client/src/main/scala/org/apache/celeborn/client/WorkerStatusTracker.scala	Adds endpoint-ready worker pool state and basic mutators/accessors.
client/src/main/scala/org/apache/celeborn/client/LifecycleManager.scala	Adds time-throttled refresh + endpoint sync logic and RequestWorkers RPC client path.
client/src/main/scala/org/apache/celeborn/client/ChangePartitionManager.scala	Switches candidate selection to the refreshed endpoint-ready pool + snapshot fallback.

Suppressed comments (1)

tests/spark-it/src/test/scala/org/apache/celeborn/tests/client/ChangePartitionManagerUpdateWorkersSuite.scala:211

• eventually(..., interval(0.milliseconds)) creates a tight retry loop that can busy-spin the CPU and hammer RPCs, making this IT test flaky under load/CI. Use a small non-zero polling interval (for example 100ms) to reduce churn while still keeping the test responsive.

    eventually(timeout(10.seconds), interval(0.milliseconds)) {

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Pull request overview

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

Suppressed comments (1)

docs/configuration/master.md:88

• The description claims “Workers already present in a shuffle snapshot are not counted against this limit”, but the RequestWorkers RPC (and Master.handleRequestWorkers) has no shuffleId/snapshot context, and the client request only excludes failed workers (LifecycleManager.requestMasterRequestWorkersWithRetry), so snapshot workers can still be returned and therefore do count toward the response limit. Please update the config docs to match actual behavior (or adjust the protocol/client to make the statement true).

| celeborn.master.splitSlot.assign.maxWorkers | 500 | false | Maximum workers returned by each dynamic candidate refresh. The request limit is the smaller positive value of this setting and `celeborn.client.slot.assign.maxWorkers`. For replicated shuffle, an effective limit of one is raised to two. Workers already present in a shuffle snapshot are not counted against this limit. | 0.7.0 |  | 

[Kalvin2077]

@RexXiong @SteNicholas
PING.