Malachi.
View Source
Tracks the members of consumer groups and assigns each topic's ranges across them (via
Malachi.Consumer.Assignment), so members consume in parallel: one range per member.
A group is keyed by {group, topic}. Members join, then heartbeat to stay alive; a member that
stops heartbeating for session_ms is evicted (a dead consumer). Rebalancing is eager: any
membership change: join, leave, eviction - or a change in the topic's ranges recomputes the whole
assignment and bumps a generation counter; a member re-reads its assignment on the next heartbeat and
sees the new generation, its signal to take over its (possibly changed) ranges. Because the assignment is
deterministic and sticky, a rebalance moves few ranges.
This is the coordinator logic as a single GenServer, reached through seams (:clock, :ranges_fun)
so it is testable without a cluster; which node coordinates which group (or replicating the membership)
is a separate wiring concern. Member state is soft: a coordinator restart just makes members re-join.
Options
:ranges_fun-(topic -> [range_id]), the topic's current ranges (defaultfn _ -> [] end):clock-(-> integer_ms)monotonic clock (defaultSystem.monotonic_time(:millisecond)):session_ms- evict a member silent this long (default 30_000):tick_ms- reconcile interval: evict + rebalance (default 10_000):name- optional registered name
Summary
Functions
Reads member's current assignment without refreshing its session.
Returns a specification to start this module under a supervisor.
Refreshes member's session and returns its current {:ok, generation, ranges}, or
{:error, :unknown_member} if it was evicted (the caller must join/4 again).
Joins member to {group, topic}, triggering a rebalance. Returns {:ok, generation, ranges}, or
{:error, :not_owner} if this node no longer owns the topic's coordination (the caller re-resolves).
Removes member from {group, topic}, triggering a rebalance.
The fetch-path entry point (implicit membership): registers member if new (a rebalance) or just
refreshes its session if known (no rebalance), and returns its {:ok, generation, ranges}. Called on
every fetch, so a member stays alive by fetching and needs no separate heartbeat. Returns
{:error, :not_owner} if this node no longer owns the topic's coordination (the caller re-resolves).
Runs one reconcile (evict timed-out members + rebalance) immediately. Also fired on the timer.
Types
@type member() :: term()
@type reply() :: {:ok, non_neg_integer(), [term()]} | {:error, :unknown_member | :not_owner}
Functions
@spec assignment(GenServer.server(), term(), term(), member()) :: reply()
Reads member's current assignment without refreshing its session.
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec heartbeat(GenServer.server(), term(), term(), member()) :: reply()
Refreshes member's session and returns its current {:ok, generation, ranges}, or
{:error, :unknown_member} if it was evicted (the caller must join/4 again).
@spec join(GenServer.server(), term(), term(), member()) :: reply()
Joins member to {group, topic}, triggering a rebalance. Returns {:ok, generation, ranges}, or
{:error, :not_owner} if this node no longer owns the topic's coordination (the caller re-resolves).
@spec leave(GenServer.server(), term(), term(), member()) :: :ok
Removes member from {group, topic}, triggering a rebalance.
@spec poll(GenServer.server(), term(), term(), member()) :: reply()
The fetch-path entry point (implicit membership): registers member if new (a rebalance) or just
refreshes its session if known (no rebalance), and returns its {:ok, generation, ranges}. Called on
every fetch, so a member stays alive by fetching and needs no separate heartbeat. Returns
{:error, :not_owner} if this node no longer owns the topic's coordination (the caller re-resolves).
@spec reconcile_now(GenServer.server()) :: :ok
Runs one reconcile (evict timed-out members + rebalance) immediately. Also fired on the timer.
@spec start_link(keyword()) :: GenServer.on_start()