A process that manages membership in a Kafka consumer group.

Consumers in a consumer group coordinate with each other through a Kafka broker to distribute the work of consuming one or several topics without any overlap. This is facilitated by the Kafka client-side assignment protocol.

Any time group membership changes (a member joins or leaves the group), a Kafka broker initiates group synchronization by asking one of the group members (the leader elected by the broker) to provide partition assignments for the whole group. KafkaEx uses a round robin partition assignment algorithm by default. This can be overridden by passing a callback function in the :partition_assignment_callback option. See KafkaEx.ConsumerGroup.PartitionAssignment for details on partition assignment functions.

A KafkaEx.ConsumerGroup process is responsible for:

1. Maintaining membership in a Kafka consumer group.
2. Determining partition assignments if elected as the group leader.
3. Launching and terminating KafkaEx.GenConsumer processes based on its assigned partitions.

To use a KafkaEx.ConsumerGroup, a developer must define a module that implements the KafkaEx.GenConsumer behaviour and start a KafkaEx.ConsumerGroup configured to use that module.

The api versions of some of the underlying messages can be specified in the :api_versions option. Note that these will be ignored (api version 0 used) unless you have kafka_version: "kayrock" set in the KafkaEx application config. The following versions can be specified:

• :fetch - Fetch requests - use v2+ for newer versions of Kafka
• :offset_fetch - Offset fetch requests - use v1+ for offsets stored in Kafka (as opposed to zookeeper)
• :offset_commit - Offset commit requests - use v1+ to store offsets in Kafka (as opposed to zookeeper)

example

Example

Suppose we want to consume from a topic called "example_topic" with a consumer group named "example_group" and we have a KafkaEx.GenConsumer implementation called ExampleGenConsumer (see the KafkaEx.GenConsumer documentation). We could start a consumer group in our application's supervision tree as follows:

defmodule MyApp do
  use Application

  def start(_type, _args) do
    consumer_group_opts = [
      # setting for the ConsumerGroup
      heartbeat_interval: 1_000,
      # this setting will be forwarded to the GenConsumer
      commit_interval: 1_000
    ]

    gen_consumer_impl = ExampleGenConsumer
    consumer_group_name = "example_group"
    topic_names = ["example_topic"]

    children = [
      # ... other children
      %{
        id: KafkaEx.ConsumerGroup,
        start: {
          KafkaEx.ConsumerGroup,
          :start_link,
          [gen_consumer_impl, consumer_group_name, topic_names, consumer_group_opts]
        }
      }
      # ... other children
    ]

    Supervisor.start_link(children, strategy: :one_for_one)
  end
end

Note It is not necessary for the Elixir nodes in a consumer group to be connected (i.e., using distributed Erlang methods). The coordination of group consumers is mediated by the broker.

See start_link/4 for configuration details.