[CELEBORN-2429][CIP-22] Batch worker/application heartbeats into aggregated raft log entries - #3810
[CELEBORN-2429][CIP-22] Batch worker/application heartbeats into aggregated raft log entries#3810yew1eb wants to merge 2 commits into
Conversation
03d16c6 to
b1c0f14
Compare
f4f3e24 to
4ece3b1
Compare
…egated raft log entries Aggregate heartbeats on the leader over a short time window (default 1s) into one BatchHeartbeat raft entry, cutting raft write volume by ~100x at peak. Off by default: celeborn.master.ha.heartbeat.batch.enabled.
| } | ||
|
|
||
| public void stop() { | ||
| flushExecutor.shutdownNow(); |
There was a problem hiding this comment.
There is no final flush before the master stops. Those heartbeats are always lost. Should we try to flush once before shut down.
There was a problem hiding this comment.
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.
| .setRequestId(MasterClient.genRequestId()) | ||
| .setBatchHeartbeatRequest( | ||
| ResourceProtos.BatchHeartbeatRequest.newBuilder() | ||
| .addAllWorkerHeartbeats(drainedWorkers) |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
…pping HeartbeatAggregator
Codecov Report✅ All modified and coverable lines are covered by tests. 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. 🚀 New features to boost your workflow:
|
2 participants
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
BatchHeartbeatraft log entry, instead of one entry per heartbeat.HeartbeatAggregatoron 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.BatchHeartbeatRequest/PbBatchHeartbeatRequest(Type.BatchHeartbeat = 31);MetaHandlerexpands the batch and applies each child heartbeat through the samehandleWorkerHeartbeat/handleAppHeartbeathelpers as the single-entry path.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 withRunningApplicationCount(~50 -> 1000). A follower 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?
Does this PR introduce any user-facing change?
How was this patch tested?
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 theResourceRequest→PbMetaRequestwire round-trip.