View Source KafkaEx.Consumer.ConsumerGroup.Assignment (kafka_ex v1.1.0)

The group leader's partition-assignment computation, split out of ConsumerGroup.Manager: fetch the assignable partitions from cluster metadata (retrying UNKNOWN_TOPIC_OR_PARTITION), run the assignment strategy over the members, and translate to/from the SyncGroup wire shape.

Run only by the group leader, kept separate from the coordinator FSM.

Summary

Functions

Topic/partition tuples the leader can assign, from cluster metadata. Retries UNKNOWN_TOPIC_OR_PARTITION with exponential backoff (Java-client pattern); after 5 attempts it assigns whatever topics were found.

Run the assignment strategy callback over members/partitions (the leader's job) and pack each member's result into the SyncGroup wire shape [{member_id, [{topic, [partition]}]}].

Translate leader assignments to the protocol-agnostic SyncGroup request shape.

Flatten a SyncGroup response's partition_assignments into [{topic, partition}].

Functions

Link to this function

assignable_partitions(client, topics, group_name)

View Source
@spec assignable_partitions(pid(), [binary()], binary()) :: [{binary(), integer()}]

Topic/partition tuples the leader can assign, from cluster metadata. Retries UNKNOWN_TOPIC_OR_PARTITION with exponential backoff (Java-client pattern); after 5 attempts it assigns whatever topics were found.

Runs synchronously in the Manager callback and :timer.sleeps between retries, briefly blocking heartbeats/shutdown — removed by the non-blocking-Manager follow-up.

Link to this function

for_members(members, partitions, callback)

View Source
@spec for_members([binary()], [{binary(), integer()}], function()) :: [
  {binary(), list()}
]

Run the assignment strategy callback over members/partitions (the leader's job) and pack each member's result into the SyncGroup wire shape [{member_id, [{topic, [partition]}]}].

Link to this function

format_for_sync_group(assignments)

View Source
@spec format_for_sync_group([{binary(), list()}]) :: [map()]

Translate leader assignments to the protocol-agnostic SyncGroup request shape.

Link to this function

from_sync_response(partition_assignments)

View Source
@spec from_sync_response([%{topic: binary(), partitions: [integer()]}]) :: [
  {binary(), integer()}
]

Flatten a SyncGroup response's partition_assignments into [{topic, partition}].