[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 1 commit 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.
| .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.
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.