Package io.krabka.streams.coordination

package io.krabka.streams.coordination
Leader election, leases, and fencing tokens for one role.

A control plane that dispatches commands to external devices needs two properties from its log. A command is durable in a quorum before the dispatch, and a deposed leader cannot dispatch at all. The second property is the hard one. A leader that loses its lease does not know that it lost the lease, so it cannot fence itself. Something outside the leader must refuse its writes.

The epoch is the safety mechanism and the lease is not

The leadership epoch is the producer epoch that Kafka's transaction coordinator mints for transactional.id = <role>. That one decision supplies every property this package promises:

  • The coordinator writes the epoch to __transaction_state, which is replicated, so the quorum mints the epoch.
  • The coordinator advances the epoch on every InitProducerId call and reuses no value, so the epoch grows.
  • The broker rejects a write that carries a superseded epoch. It answers INVALID_PRODUCER_EPOCH or PRODUCER_FENCED. The cluster fences a deposed leader, and the leader does not fence itself.
  • DescribeTransactions reports the current producer id and producer epoch of a transactional id. Any process calls it, so a third party checks the authority of a writer with one request and no membership. See LeadershipStatus.

The lease adds no safety of its own. It is a liveness and anti-flap device. It tells a standby when to stop waiting for a quiet holder. A wrong lease makes a failover early or late. A wrong lease never makes two writers authoritative. Read that inversion against a hand-rolled design, where the lease deadline decides who may write. This package never asks the lease that question.

The fencing token is a pair

The producer epoch is a short and it wraps. Kafka handles the exhaustion with a fresh producer id and an epoch of zero. An implementation that compares the epoch alone accepts a stale writer after about 32000 leadership changes. FencingToken holds the pair and compares the producer id first.

Rank comes from the log

A candidate appends a registration record to __coordination_state. The offset of that record is the join sequence of the candidate, and log compaction keeps the offset of every record it retains. RoleState folds that walk, and the roster it builds is in offset order. A recovered node registers again, takes a higher offset, and lands at the tail. That is the no-failback rule, and it needs no counter and no coordinator. A rank from a configuration file would put the recovered node at the front, and the node would then preempt the member that replaced it.

The layers

Clock assumptions

The lease deadline is a wall-clock instant. The holder writes it, and a challenger compares it against its own clock. Skew between the two clocks moves the failover time by the size of the skew. Skew does not affect safety, for the reason the first section gives. An operator that needs a bound on the failover time needs a bound on the skew.

Example

Role role = Role.of("controller");
MemberId me = MemberId.of("node-1");
try (CoordinationClient client = new CoordinationClient(transport);
        Leadership leadership = client.acquire(role, me, Duration.ofMinutes(1))) {
    while (running) {
        if (leadership.renewDue()) {
            leadership.renew();
        }
        dispatch(leadership.token());
    }
} catch (FencedException lost) {
    controller.stop();
}