Crate krabka_client_coordination

Expand description

Leader election, leases, and fencing tokens for Apache Kafka clients.

This crate gives an active or standby process a way to elect one leader per role and to prove that leadership on every guarded write. It builds the primitives on Kafka itself, so a cluster needs no second consensus service. Per-role state lives in the compacted internal topic COORDINATION_STATE_TOPIC.

The leadership epoch is the producer epoch. Kafka’s transaction coordinator mints that epoch when a member calls InitProducerId for the transactional.id of the role. The quorum picks the value, so the epoch is monotonic, and every broker enforces it. A write that carries an older epoch fails with a fenced error. The lease adds no safety of its own. It is an anti-flap device: a standby waits for the deadline of the holder to pass before it challenges, so a short pause does not move the role.

§Key Types

Role and MemberId — the validated names of a role and of one member that competes for it.
FencingToken — the quorum-minted proof that a member holds a role.
CoordinationKey and CoordinationRecord — the decoded key and value of one record of COORDINATION_STATE_TOPIC.
encode_key, decode_key, encode_value, and decode_value — the codec for the frozen record layouts.

§Quick start

The record codec is the part of the crate that exists now. It encodes and decodes the frozen layouts that krabka-streams-java and krabka-streams-go also implement.

use krabka_client_coordination::{
    CoordinationKey, CoordinationRecord, FencingToken, Lease, MemberId, Role, decode_key,
    decode_value, encode_key, encode_value,
};

let key = CoordinationKey::lease(Role::new("controller")?);
let record = CoordinationRecord::Lease(Lease {
    member: MemberId::new("node-1")?,
    token: FencingToken::new(4242, 7)?,
    granted_at: 1_700_000_000_000,
    deadline: 1_700_000_030_000,
});

// Produce these two buffers to `__coordination_state`. A `None` value is a
// tombstone, and it clears the lease.
let key_bytes = encode_key(&key);
let value_bytes = encode_value(&record);

// Read them back on the consumer side.
let decoded_key = decode_key(&key_bytes)?;
let decoded = decode_value(decoded_key.kind, value_bytes.as_deref())?;
let CoordinationRecord::Lease(lease) = decoded else {
    unreachable!("the record is a lease")
};
println!(
    "{} holds {} with token {} until {}",
    lease.member, decoded_key.role, lease.token, lease.deadline
);

§Scope and boundaries

This crate owns the coordination record format and the client API built on it. It does not change the broker, and it needs no broker-side feature. A caller supplies its own producer and consumer, and this crate supplies the succession rules, the lease clock, and the codec below.

Structs§

BrokerTransport: The CoordinationTransport that talks to a Kafka cluster.
CoordinationKey: The decoded key of a coordination state record.
FencingToken: The quorum-minted proof that one member holds a role.
Lease: The member that holds a role now, and the time its claim ends.
LeaseConfig: The lease timings of one role.
LeaseTiming: The lease clock of one lease record under one policy.
MemberId: The identity of one member that competes for a role.
Registration: One member of a role announces that it is available.
Role: The name of a coordination role, such as the name of an active controller.
RoleState: The state that a reader builds from the records of one role.
RoleStateBuilder: Folds the records of one role into a RoleState.
RosterEntry: One member of the roster of a role.
SystemClock: The clock of the host, read through SystemTime.
TestClock: A clock that a test moves by hand.

Enums§

CoordinationError: A fault that the coordination client reports.
CoordinationRecord: One decoded value of COORDINATION_STATE_TOPIC.
Decision: What one member does about a role right now.
RecordError: A failure to build or to decode a coordination state record.
RecordKind: Which record of COORDINATION_STATE_TOPIC a key names.

Constants§

COORDINATION_STATE_TOPIC: The compacted internal topic that holds the coordination state of a cluster.
DEFAULT_CHALLENGE_STAGGER: The default extra delay that one rank of succession adds.
DEFAULT_COORDINATION_CLIENT_ID: The client id that the transport gives every connection it opens.
DEFAULT_COORDINATION_FETCH_MAX_WAIT: The time that one fetch of the coordination topic waits for records.
DEFAULT_COORDINATION_FETCH_PARTITION_MAX: The byte limit that the transport puts on one partition fetch.
DEFAULT_COORDINATION_PARTITIONS: The number of partitions that the transport gives the coordination topic when it creates it.
DEFAULT_COORDINATION_REPLICATION: The replication factor that the transport gives the coordination topic when it creates it.
DEFAULT_COORDINATION_REQUEST_TIMEOUT: The deadline that the transport puts on one request.
DEFAULT_LEASE_DURATION: The default extent of a lease.
DEFAULT_RENEW_INTERVAL: The default gap between two renewals by the holder.
MAX_MEMBER_ID_LEN: The maximum length of a MemberId, in bytes.
MAX_ROLE_LEN: The maximum length of a Role, in bytes.

Traits§

Clock: A source of the current time.
CoordinationTransport: The cluster operations that the coordination client performs.

Functions§

decode_key: Decodes a key of COORDINATION_STATE_TOPIC.
decode_value: Decodes a value of COORDINATION_STATE_TOPIC under the kind of its key.
encode_key: Encodes a key of COORDINATION_STATE_TOPIC.
encode_value: Encodes a value of COORDINATION_STATE_TOPIC.
evaluate: Decides what me does about the role of state at now_millis.
role_partition: The partition that holds every record of role.

Type Aliases§

RoleRecords: The records of one role’s partition, in offset order.