[CELEBORN-2429][CIP-22] Batch worker/application heartbeats into aggregated raft log entries

[yew1eb]

What changes were proposed in this pull request?

In HA mode, aggregate worker/application heartbeats on the master leader over a short time window (default 1s) and replicate them as ONE BatchHeartbeat raft log entry, instead of one entry per heartbeat.

• New HeartbeatAggregator on the leader: pending heartbeats are kept in per-worker/per-app maps (newest wins, buffer intrinsically bounded), a scheduled thread flushes one batch per window; heartbeat RPCs are enqueued asynchronously and replied immediately from leader-local memory.
• New internal protobuf messages BatchHeartbeatRequest / PbBatchHeartbeatRequest (Type.BatchHeartbeat = 31); MetaHandler expands the batch and applies each child heartbeat through the same handleWorkerHeartbeat / handleAppHeartbeat helpers as the single-entry path.
• Fully additive and off by default:
  celeborn.master.ha.heartbeat.batch.enabled (false),
  celeborn.master.ha.heartbeat.batch.interval (1s).

Design doc (CIP-22 Batched Heartbeat Raft Writes): https://docs.google.com/document/d/1YXDCjk_kR_5jyRGmUz1DqImjnAtsr7oFEwpYKK1NxAE/edit?usp=sharing

Why are the changes needed?

Heartbeats are the most frequent metadata writes: A/10 + W/30 raft entries/s for A apps and W workers. On one of our production clusters at peak (900+ running apps + 200 workers, ~97 heartbeat entries/s), follower apply lag (RatisApplyCompletedIndexDiff) grows in lockstep with RunningApplicationCount (~50 -> 1000). A follower master flame graph shows the cost is the per-entry raft pipeline (appendEntries ~35%, log flush/fsync ~20%), not apply compute (~7.7%).

Heartbeats are periodic, self-healing, timeout-tolerant, and their replies are built from leader-local memory — they do not need a synchronous raft commit each. Batching cuts raft write volume by ~100x at peak.

Does this PR resolve a correctness bug?

• Yes

Does this PR introduce any user-facing change?

• Yes

How was this patch tested?

• New HeartbeatAggregatorSuiteJ (single-node raft, full offer → submit → replicate → apply chain): N offers produce far fewer entries; empty windows produce none; duplicate keys within one window collapse to the newest heartbeat.
• MasterStateMachineSuiteJ#testBatchHeartbeat: batched entry applies correctly and survives the ResourceRequest → PbMetaRequest wire round-trip.
• Existing master suites pass.

[yew1eb]

Ping @pan3793 @cxzl25 @RexXiong @SteNicholas @zaynt4606

[AmandeepSingh285]

There is no final flush before the master stops. Those heartbeats are always lost. Should we try to flush once before shut down.

[yew1eb]

Good catch. Master.stop() stops the aggregator before the raft server and before leadereship transfer, so a final flush can still commit. Added a best-effort flushSafely() in stop(): on the leader it commits the pending batch synchronously; on a non-leader it's a no-op.

[AmandeepSingh285]

If the active leader dies before the batched worker heartbeat is committed to Ratis, the new leader keeps the last applied worker snapshot. Load-aware slot allocation then ranks disks from stale flush/fetch times, usable space, and active slots, so traffic can keep landing on workers that have since become hot or unhealthy.
highWorkload is applied only when that heartbeat is applied. If the lost heartbeat was the one that marked the worker overloaded, the new leader will still treat it as available and offer new slots until the next heartbeat (~30s).
The same gap exists for decommission: the worker may already be draining from an event it received, but the status it reported in the lost heartbeat never reaches the new leader. Until the next heartbeat applies, the master can still give applications slots on a worker that is leaving the cluster.

The impact of losing a worker/app heartbeat after this change could be significantly higher, as losing the subsequent heartbeat would further extend the period during which the leader operates on stale worker state.

[yew1eb]

This is a valid concern: if the leader dies before the batch commits, the new leader loses up to one flush window of heartbeats for all workers/apps. That said, the incremental risk is bounded by the flush interval T (default 1s), not the heartbeat period: batched or not, a lost heartbeat is only refreshed by that worker's next heartbeat (30s), so the same stale gap already exists today whenever a heartbeat is lost in a failover. With the final flush added above, graceful shutdowns (rolling restart, the common case in production) lose nothing regardless of T. After a leader change, the extended worker/app timeout deadlines prevent false WorkerLost/AppLost during the convergence window, and workers re-report their highWorkload/decommission status in the next heartbeat, so the state self-heals the same way the non-batch path relies on.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 58.55%. Comparing base (90954f7) to head (6a329a5).
⚠️ Report is 24 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #3810      +/-   ##
============================================
- Coverage     58.74%   58.55%   -0.18%     
  Complexity      231      231              
============================================
  Files           399      399              
  Lines         28058    28080      +22     
  Branches       2740     2745       +5     
============================================
- Hits          16479    16440      -39     
- Misses        10380    10437      +57     
- Partials       1199     1203       +4     

☔ 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.

[AmandeepSingh285]

From the underlying implementation, submitRequest() appears to be synchronous since it waits on

raftClientReply = server.submitClientRequestAsync(raftClientRequest).get();

If the Ratis submission takes a long time, the single flush thread will be blocked and subsequent flushes will be delayed. During this time, new worker/app heartbeats will continue accumulating in the pending maps.
Should we consider putting an upper bound on the number of pending heartbeat entries to prevent unbounded memory growth in case the Ratis call is blocked for an extended period? If the limit is reached, we could drop subsequent heartbeats and emit a warning log so that this condition is visible.

[yew1eb]

You're right that submitRequest() is synchronous, so a blocked Ratis call would stall the single flush thread and delay subsequent flushes. But an upper bound on pending entries shouldn't be necessary:

1. Entries don't accumulate while the thread is blocked. The maps are keyed by worker/app id with newest-wins semantics, and flush() drains and clears them under the lock before submitRequest — so heartbeats arriving during a blocked submit just overwrite into fresh maps, bounded by #workers + #apps, the same order as the master's own metadata.
2. This blocking is pre-existing, with a larger blast radius on the non-batch path. There, every heartbeat synchronously calls the same submitRequest on an RPC handler thread — a slow Ratis can stall many RPC threads at once, one per in-flight heartbeat. Batching reduces this to a single dedicated flush thread, and collapses per-heartbeat raft submions into one per flush window.
3. The impact stays small: a delayed flush means at most a few extra seconds of stale heartbeat state — far below the 120s/300s timeouts, self-heals next window — and the existingn log already fires when a submit exceeds the flush interval, so the condition is visible.