docs: KRaft protocol reference for M1.1 spec writing
This commit is contained in:
Scott Scoble
parent
6e770de885
commit
1d536136b8
1 changed files with 318 additions and 0 deletions
318
docs/kraft-protocol-reference.md
Normal file
318
docs/kraft-protocol-reference.md
Normal file
|
|
@ -0,0 +1,318 @@
|
|||
# Kafka KRaft Protocol Reference
|
||||
|
||||
> Source material for writing `examples/KafkaKRaft.speck` (speckl#22)
|
||||
|
||||
## Scope
|
||||
|
||||
This reference covers the **consensus core** of KRaft — the state machine for leader election, log replication, and commit advancement. It does NOT cover:
|
||||
|
||||
- Full broker functionality (topic creation, partition assignment, consumer group coordination)
|
||||
- Data plane (produce/fetch request handling)
|
||||
- Controller-specific logic (metadata quarums, configuration management)
|
||||
- Network transport (this is a logical model, not a network protocol spec)
|
||||
|
||||
The boundary: if it touches `currentTerm`, `votedFor`, `log`, `commitIndex`, or `state` transitions, it's in scope.
|
||||
|
||||
## Sources
|
||||
|
||||
1. **Raft TLA+ Specification** — Diego Ongaro, 2014. The canonical formal spec. Available at: `github.com/ongardie/raft.tla`
|
||||
2. **"In Search of an Understandable Consensus Algorithm"** — Ongaro & Ousterhout, 2014 (USENIX ATC). Includes safety proof.
|
||||
3. **KIP-500: Next Generation Client Protocol** — KRaft design document. URL changes; search Apache Kafka JIRA for KIP-500.
|
||||
4. **KIP-642: Dynamic Configuration in KRaft** — Configuration updates in KRaft mode.
|
||||
5. **KIP-830: KRaft Metadata Quorum** — The actual metadata quorum implementation.
|
||||
6. **Apache Kafka source** — `kafka/raft/` package in the Apache Kafka repository.
|
||||
|
||||
## Raft State Machine (from TLA+ Spec)
|
||||
|
||||
### State Variables
|
||||
|
||||
Per server (`i ∈ Server`):
|
||||
|
||||
| Variable | Type | Description |
|
||||
|----------|------|-------------|
|
||||
| `currentTerm` | Nat | Current term number (monotonically increasing) |
|
||||
| `state` | {Follower, Candidate, Leader} | Server role |
|
||||
| `votedFor` | Server ∪ {Nil} | Candidate voted for in current term, or Nil |
|
||||
| `log` | Seq(LogEntry) | Sequence of log entries (index = position) |
|
||||
| `commitIndex` | Nat | Index of highest log entry known to be committed |
|
||||
| `nextIndex` | Server → Nat | (Leader only) Next log index to send to each follower |
|
||||
| `matchIndex` | Server → Nat | (Leader only) Highest log index replicated on each follower |
|
||||
| `votesResponded` | Set(Server) | (Candidate only) Servers that responded to RequestVote |
|
||||
| `votesGranted` | Set(Server) | (Candidate only) Servers that granted vote |
|
||||
|
||||
Global:
|
||||
|
||||
| Variable | Type | Description |
|
||||
|----------|------|-------------|
|
||||
| `messages` | Bag(Message) | In-flight messages (requests/responses) |
|
||||
|
||||
### LogEntry Type
|
||||
|
||||
```
|
||||
LogEntry == { term: Nat, value: Value }
|
||||
```
|
||||
|
||||
### Quorum
|
||||
|
||||
```
|
||||
Quorum == {i ∈ SUBSET(Server) : Cardinality(i) * 2 > Cardinality(Server)}
|
||||
```
|
||||
|
||||
Simple majority. The key property: any two quorums overlap.
|
||||
|
||||
### Initial State
|
||||
|
||||
```
|
||||
currentTerm[i] = 1
|
||||
state[i] = Follower
|
||||
votedFor[i] = Nil
|
||||
log[i] = <<>> (empty sequence)
|
||||
commitIndex[i] = 0
|
||||
```
|
||||
|
||||
### Actions (State Transitions)
|
||||
|
||||
#### 1. Timeout → Start Election
|
||||
|
||||
**Preconditions:**
|
||||
- `state[i] ∈ {Follower, Candidate}`
|
||||
|
||||
**Effects:**
|
||||
- `state[i] ← Candidate`
|
||||
- `currentTerm[i] ← currentTerm[i] + 1`
|
||||
- `votedFor[i] ← Nil`
|
||||
- `votesResponded[i] ← {}`
|
||||
- `votesGranted[i] ← {}`
|
||||
|
||||
#### 2. RequestVote Request (Candidate → Peer)
|
||||
|
||||
**Preconditions:**
|
||||
- `state[i] = Candidate`
|
||||
- `j ∉ votesResponded[i]`
|
||||
|
||||
**Effects:**
|
||||
- Send `RequestVoteRequest` with `{mterm: currentTerm[i], mlastLogTerm: LastTerm(log[i]), mlastLogIndex: Len(log[i]), msource: i, mdest: j}`
|
||||
|
||||
#### 3. HandleRequestVoteRequest (Peer → Candidate)
|
||||
|
||||
**Preconditions:**
|
||||
- `m.mterm ≤ currentTerm[i]`
|
||||
|
||||
**Log check (`logOk`):**
|
||||
```
|
||||
logOk ≡ m.mlastLogTerm > LastTerm(log[i])
|
||||
∨ (m.mlastLogTerm = LastTerm(log[i]) ∧ m.mlastLogIndex ≥ Len(log[i]))
|
||||
```
|
||||
|
||||
**Grant conditions:**
|
||||
```
|
||||
grant ≡ m.mterm = currentTerm[i] ∧ logOk ∧ votedFor[i] ∈ {Nil, j}
|
||||
```
|
||||
|
||||
**Effects (if grant):**
|
||||
- `votedFor[i] ← j`
|
||||
|
||||
**Effects (always):**
|
||||
- Reply with `RequestVoteResponse{mterm: currentTerm[i], mvoteGranted: grant}`
|
||||
|
||||
#### 4. HandleRequestVoteResponse (Candidate receives vote)
|
||||
|
||||
**Preconditions:**
|
||||
- `m.mterm = currentTerm[i]`
|
||||
|
||||
**Effects:**
|
||||
- `votesResponded[i] ← votesResponded[i] ∪ {j}`
|
||||
- If `m.mvoteGranted`: `votesGranted[i] ← votesGranted[i] ∪ {j}`
|
||||
|
||||
#### 5. BecomeLeader (Candidate → Leader)
|
||||
|
||||
**Preconditions:**
|
||||
- `state[i] = Candidate`
|
||||
- `votesGranted[i] ∈ Quorum`
|
||||
|
||||
**Effects:**
|
||||
- `state[i] ← Leader`
|
||||
- `nextIndex[i] ← [j ∈ Server → Len(log[i]) + 1]` for all j
|
||||
- `matchIndex[i] ← [j ∈ Server → 0]` for all j
|
||||
|
||||
#### 6. AppendEntries Request (Leader → Follower)
|
||||
|
||||
**Preconditions:**
|
||||
- `i ≠ j`
|
||||
- `state[i] = Leader`
|
||||
|
||||
**Effects:**
|
||||
- Send `AppendEntriesRequest` with entries from `nextIndex[i][j]` to `min(Len(log[i]), nextIndex[i][j])` (at most 1 entry per message in the spec)
|
||||
- Include `mprevLogIndex = nextIndex[i][j] - 1`, `mprevLogTerm`, `mcommitIndex`
|
||||
|
||||
#### 7. HandleAppendEntriesRequest (Follower receives entries)
|
||||
|
||||
**Preconditions:**
|
||||
- `m.mterm ≤ currentTerm[i]`
|
||||
|
||||
**Log consistency check (`logOk`):**
|
||||
```
|
||||
logOk ≡ m.mprevLogIndex = 0
|
||||
∨ (m.mprevLogIndex ≤ Len(log[i]) ∧ m.mprevLogTerm = log[i][m.mprevLogIndex].term)
|
||||
```
|
||||
|
||||
**Effects (if logOk AND m.mterm ≥ currentTerm[i]):**
|
||||
- Reset to Follower if Candidate
|
||||
- Update `currentTerm[i] ← m.mterm` if needed
|
||||
- Append/conflict resolution: if `m.mprevLogIndex > 0`, entries after `mprevLogIndex` that conflict are truncated
|
||||
- `commitIndex[i] ← min(m.mcommitIndex, Len(log[i]))`
|
||||
- Reply `AppendEntriesResponse{msuccess: TRUE, mmatchIndex: ...}`
|
||||
|
||||
**Effects (if !logOk OR m.mterm < currentTerm[i]):**
|
||||
- Reply `AppendEntriesResponse{msuccess: FALSE, mmatchIndex: 0}`
|
||||
|
||||
#### 8. AdvanceCommitIndex (Leader)
|
||||
|
||||
**Preconditions:**
|
||||
- `state[i] = Leader`
|
||||
|
||||
**Effects:**
|
||||
- Find highest `N` such that `N > commitIndex[i]` and a quorum of `matchIndex[i][k] ≥ N`
|
||||
- Only commit entries from current term: `log[i][N].term = currentTerm[i]`
|
||||
- `commitIndex[i] ← N`
|
||||
|
||||
#### 9. ClientRequest (Leader appends entry)
|
||||
|
||||
**Preconditions:**
|
||||
- `state[i] = Leader`
|
||||
|
||||
**Effects:**
|
||||
- Append `[term: currentTerm[i], value: v]` to `log[i]`
|
||||
|
||||
#### 10. Restart (Server crash recovery)
|
||||
|
||||
**Effects:**
|
||||
- `state[i] ← Follower`
|
||||
- Reset volatile state (`votesResponded`, `votesGranted`, `nextIndex`, `matchIndex`, `commitIndex`)
|
||||
- Preserve persistent state (`currentTerm`, `votedFor`, `log`)
|
||||
|
||||
### Safety Invariants (from Raft paper)
|
||||
|
||||
These are the invariants we need to prove in SpeckDL:
|
||||
|
||||
1. **Election Safety:** At most one leader per term.
|
||||
```
|
||||
∀ i, j ∈ Server: state[i] = Leader ∧ state[j] = Leader ∧ currentTerm[i] = currentTerm[j] → i = j
|
||||
```
|
||||
|
||||
2. **Leader Completeness:** If a log entry is committed in term T, then all leaders for terms > T have that entry.
|
||||
```
|
||||
This is the core invariant. Proved via the "Leader Completeness Property" in the Raft paper.
|
||||
```
|
||||
|
||||
3. **Log Matching:** If two entries in different logs have the same index and term, then all preceding entries are identical.
|
||||
```
|
||||
∀ i, j ∈ Server, idx:
|
||||
log[i][idx].term = log[j][idx].term → log[i][1..idx] = log[j][1..idx]
|
||||
```
|
||||
|
||||
4. **Leader Append-Only:** A leader never overwrites or deletes entries in its log.
|
||||
```
|
||||
∀ i: state[i] = Leader → log[i] only grows (entries appended, never removed)
|
||||
```
|
||||
|
||||
5. **Term Monotonicity:** Current term never decreases.
|
||||
```
|
||||
∀ i: currentTerm' [i] ≥ currentTerm[i]
|
||||
```
|
||||
|
||||
6. **Commit Safety:** An entry is committed only if a quorum has replicated it.
|
||||
```
|
||||
commitIndex[i] advances only when a quorum of matchIndex[i][k] ≥ N
|
||||
```
|
||||
|
||||
## KRaft Extensions Over Vanilla Raft
|
||||
|
||||
KRaft is Kafka's adaptation of Raft for metadata management. Key differences:
|
||||
|
||||
1. **No ZooKeeper:** KRaft replaces the ZooKeeper-based metadata quorum with a Raft-based one.
|
||||
2. **Metadata log:** The replicated log contains metadata records (topic configs, partition assignments), not user data.
|
||||
3. **Controller node:** KRaft has an "active controller" concept — the leader of the metadata quorum acts as the controller for the Kafka cluster.
|
||||
4. **Snapshotting:** KRaft supports snapshot-based log compaction (metadata snapshots).
|
||||
5. **Voters and observers:** Not all nodes are voters; some are observer nodes that replicate but don't vote.
|
||||
6. **Resignation:** A leader can voluntarily step down (immediate or graceful).
|
||||
|
||||
### KRaft Additional State
|
||||
|
||||
| Variable | Type | Description |
|
||||
|----------|------|-------------|
|
||||
| `isVoter` | Server → Bool | Whether this node is a voter (vs observer) |
|
||||
| `isLeader` | Server → Bool | Cached: `state[i] = Leader` |
|
||||
| `highWatermark` | Nat | Similar to commitIndex; offset of last stable metadata record |
|
||||
|
||||
### KRaft Additional Actions
|
||||
|
||||
- **Resign:** Leader steps down gracefully
|
||||
- **BecomeObserver:** Voter transitions to observer (loses vote)
|
||||
- **BecomeVoter:** Observer transitions to voter (gains vote)
|
||||
|
||||
## Known Verification Gaps / Interesting Targets
|
||||
|
||||
1. **Election safety under network partitions:** Raft's election safety depends on the quorum overlap property. In practice, network partitions and message reordering can create edge cases.
|
||||
|
||||
2. **Log truncation after leader change:** When a new leader overwrites follower logs that conflict, the truncation must be atomic. This is a known source of bugs in Raft implementations.
|
||||
|
||||
3. **Stale leader problem:** After a network partition heals, a stale leader may attempt to commit entries from an old term. AdvanceCommitIndex prevents this by only committing entries from the current term.
|
||||
|
||||
4. **KRaft observer consistency:** Observers don't vote but must maintain log consistency. The spec needs to ensure observers can't diverge from voters.
|
||||
|
||||
5. **Graceful shutdown / resignation race:** If a leader resigns while a client request is in flight, the entry may or may not be committed. This is a liveness property, not safety, but worth verifying.
|
||||
|
||||
## Producer Acknowledgement Semantics (for M1.3)
|
||||
|
||||
### acks=0 (Fire and forget)
|
||||
- Producer sends and doesn't wait for response
|
||||
- No durability guarantee
|
||||
- Invariant: **None** (messages can be lost)
|
||||
|
||||
### acks=1 (Leader acknowledgment)
|
||||
- Producer waits for leader to write to local log
|
||||
- Durability: survives leader failure, but NOT follower failure
|
||||
- Invariant: **If ack received, entry exists in leader's log**
|
||||
|
||||
### acks=all (ISR acknowledgment)
|
||||
- Producer waits for all in-sync replicas (ISR) to acknowledge
|
||||
- Durability: survives any failure that leaves a quorum
|
||||
- Invariant: **If ack received, entry is committed (quorum has replicated)**
|
||||
|
||||
### Exactly-once Semantics
|
||||
- Requires idempotent producer + transaction coordinator
|
||||
- Idempotent producer: assigns sequence numbers; broker deduplicates
|
||||
- Transaction coordinator: two-phase commit across partitions
|
||||
|
||||
### Key Invariants for Producer Acks
|
||||
|
||||
1. `acks=all` → no data loss on leader failover (bounded to quorum availability)
|
||||
2. Idempotent producer → no duplicate messages for same PID + sequence number
|
||||
3. Transaction atomicity → all partitions in a transaction commit or none do
|
||||
|
||||
## Consumer Group Rebalancing (for M1.3)
|
||||
|
||||
### States
|
||||
- Empty → Joining → Stable → Rebalancing → Stable
|
||||
|
||||
### Actions
|
||||
- JoinGroup: consumer requests to join
|
||||
- SyncGroup: consumer receives assignment
|
||||
- Heartbeat: consumer indicates liveness
|
||||
- LeaveGroup: consumer departs
|
||||
- Rebalance: triggered by membership change or topic partition change
|
||||
|
||||
### Invariants
|
||||
1. No concurrent assignments to the same partition
|
||||
2. All partitions assigned when group is stable
|
||||
3. Offset monotonicity: committed offsets never decrease
|
||||
4. No two consumers assigned the same partition simultaneously
|
||||
|
||||
## Scope Boundary
|
||||
|
||||
**In scope (M1.1):** Core Raft consensus — leader election, log replication, commit advancement, and all 6 safety invariants.
|
||||
|
||||
**Deferred to M1.3:** Producer ack semantics, consumer group rebalancing, transaction coordinator.
|
||||
|
||||
**Out of scope:** Full Kafka broker (produce/fetch handling), controller-specific logic, snapshot/recovery, network layer.
|
||||
Loading…
Add table
Add a link
Reference in a new issue