Coordinators

Active contributors: David Jacot, Andrew Schofield, Lianet Magrans, Apoorv Mittal, Lucas Brutschy

Purpose

Brokers host three replicated state machines used to coordinate clients:

• Group coordinator — consumer groups (classic + KIP-848 protocols) and Streams group protocol participants.
• Share coordinator — share groups (queue-style consumption, KIP-932).
• Transaction coordinator — exactly-once produce + consume + offset commit (KIP-98).

All three live as their own Gradle modules (group-coordinator/, share-coordinator/, transaction-coordinator/) but share a runtime in coordinator-common/ that handles event scheduling, replicated-state-machine semantics, and snapshotting onto an internal Kafka topic.

How a coordinator works

graph TD
    subgraph BrokerProcess[A broker process]
      RT[CoordinatorRuntime<br/>coordinator-common] --> SM[Per-shard state machine<br/>e.g. GroupMetadataManager]
      SM --> Snap[(In-memory state)]
      RT --> Tlog[(Internal topic partition<br/>__consumer_offsets / __share_group_state / __transaction_state)]
    end

    Cli[Client RPC] --> KA[KafkaApis] --> RT
    RT -->|read-only| Snap
    RT -->|append| Tlog
    Tlog -->|replay on partition load| SM

The runtime exposes operations like "I want to handle this RPC; let me read or mutate state for this group; persist the resulting records to the partition; respond to the caller once the records are committed". It supports scheduled writes (so multiple RPCs can be batched into one log append), read operations (no log write), and timer events (for session timeouts and rebalance deadlines).

Group coordinator (group-coordinator/)

Component	File
GroupCoordinatorService	group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java
GroupMetadataManager	group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java (most-touched file in the repo over last 90 days)
OffsetMetadataManager	group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java
Records / state machine schemas	group-coordinator/src/main/resources/common/message/ (OffsetCommitValue.json, ConsumerGroupMetadataValue.json, ...)
Membership, assignor implementations (server-side, KIP-848)	group-coordinator/.../coordinator/group/assignor/

The coordinator implements two protocols simultaneously:

• Classic protocol. Handles JoinGroup, SyncGroup, Heartbeat, LeaveGroup. Assignment is computed by a designated "leader" client; the broker just brokers messages between members.
• Consumer group protocol (KIP-848). Handles ConsumerGroupHeartbeat and ConsumerGroupDescribe. Assignment is computed on the broker by a server-side PartitionAssignor (e.g., UniformAssignor, RangeAssignor). Members poll for assignments; rebalances are incremental.

A new Streams group protocol (KIP-1071) reuses the same coordinator for Streams-specific assignment of active and standby tasks.

State persists onto the __consumer_offsets internal topic (one shard per partition). On partition leadership transfer, the coordinator replays the log to rebuild in-memory state.

Share coordinator (share-coordinator/)

KIP-932 ("share groups", queue-style consumption) introduces a second coordinator that tracks per-record acknowledgement state — at-least-once delivery, configurable redelivery on failure. Records on the __share_group_state topic carry per-record state transitions (SHARE_FETCH, ACKNOWLEDGE, RELEASE). The corresponding consumer is KafkaShareConsumer in clients/.

Component	File
ShareCoordinatorService	share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorService.java
ShareCoordinatorShard	share-coordinator/src/main/java/org/apache/kafka/coordinator/share/ShareCoordinatorShard.java
Record schemas	share-coordinator/src/main/resources/common/message/

KIP-932 was marked generally available in November 2025.

Transaction coordinator (transaction-coordinator/ + Scala glue in core/)

The transaction coordinator implements the broker side of EOS (KIP-98 and successors). For every transactional transactional.id:

1. InitProducerId issues a producer ID + epoch.
2. AddPartitionsToTxn / AddOffsetsToTxn record the partitions involved.
3. EndTxn (commit or abort) writes COMMIT / ABORT markers, which the broker then propagates to the partition logs as control records.

State is persisted on the __transaction_state topic. The Java module transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/ carries the new state machine, while some legacy code remains in core/src/main/scala/kafka/coordinator/transaction/ and is being progressively ported over.

Component	File
TransactionCoordinator	core/src/main/scala/kafka/coordinator/transaction/TransactionCoordinator.scala
TransactionStateManager	core/src/main/scala/kafka/coordinator/transaction/TransactionStateManager.scala
Java state-machine pieces	transaction-coordinator/src/main/java/org/apache/kafka/coordinator/transaction/

See features/exactly-once.md for the protocol and client side.

Shared runtime (coordinator-common/)

Type	File	Purpose
CoordinatorRuntime<S, U>	coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java	Generic runtime parameterized by shard state S and record value type U.
CoordinatorShard<U>	coordinator-common/.../runtime/CoordinatorShard.java	Per-partition state machine; implementations are GroupCoordinatorShard, ShareCoordinatorShard, TransactionCoordinatorShard.
CoordinatorEventProcessor	coordinator-common/.../runtime/CoordinatorEventProcessor.java	Single-threaded event loop per shard, ensures serialized state mutation.
CoordinatorLoader	coordinator-common/.../runtime/CoordinatorLoader.java	Replays the partition log on partition load.
CoordinatorPartitionWriter	coordinator-common/.../runtime/CoordinatorPartitionWriter.java	Wraps ReplicaManager.appendRecords for coordinators.

The runtime is what makes it cheap to add a new coordinator: implement CoordinatorShard<MyValueType>, register it as a service, write JSON schemas for your records, and the runtime handles loading, replay, scheduling, and persistence.

Sharding

Each coordinator's state is sharded by the corresponding internal topic's partitions. For consumer groups, Utils.abs(groupId.hashCode()) % numPartitions(__consumer_offsets) determines which broker (the leader of that partition) coordinates the group. Clients learn the coordinator via FindCoordinator. On leadership change of the internal topic partition, the new leader replays the partition log to rebuild state.

Entry points for modification

• Change consumer-group rebalance behavior: GroupMetadataManager.java plus the relevant assignor/ implementation. Almost all KIP-848 work lives here.
• Add a new coordinator (rare): build a CoordinatorShard<U> on top of CoordinatorRuntime, plus matching JSON schemas and a *CoordinatorService to expose it via KafkaApis.
• Change EOS semantics: TransactionCoordinator.scala, TransactionStateManager.scala, plus clients/.../producer/internals/TransactionManager.java. Almost always needs a KIP.
• Add or change a server-side assignor: implement org.apache.kafka.coordinator.group.api.assignor.ConsumerGroupPartitionAssignor (in group-coordinator-api/) and register it.

Related pages

• Modules: clients — KafkaConsumer / KafkaShareConsumer are the client side.
• Modules: core (broker) — KafkaApis dispatches to these coordinators.
• Features: Exactly-once semantics.
• Features: KIP-848 / consumer protocol — discussed under features.