A behaviour module for implementing a Kafka consumer.

A KafkaEx.GenConsumer is an Elixir process that consumes messages from Kafka. A single KafkaEx.GenConsumer process consumes from a single partition of a Kafka topic. Several KafkaEx.GenConsumer processes can be used to consume from multiple partitions or even multiple topics. Partition assignments for a group of KafkaEx.GenConsumers can be defined manually using KafkaEx.GenConsumer.Supervisor or coordinated across a cluster of nodes using KafkaEx.ConsumerGroup.

A KafkaEx.GenConsumer must implement three callbacks. Two of these will be defined with default behavior if you add use KafkaEx.GenConsumer to your module, leaving just handle_message_set/2 to be implemented. This is the recommended usage.

example

Example

The KafkaEx.GenConsumer behaviour abstracts common Kafka consumer interactions. KafkaEx.GenConsumer will take care of the details of determining a starting offset, fetching messages from a Kafka broker, and committing offsets for consumed messages. Developers are only required to implement handle_message_set/2 to process messages.

The following is a minimal example that logs each message as it's consumed:

defmodule ExampleGenConsumer do
  use KafkaEx.GenConsumer

  alias KafkaEx.Protocol.Fetch.Message

  require Logger

  # note - messages are delivered in batches
  def handle_message_set(message_set, state) do
    for %Message{value: message} <- message_set do
      Logger.debug(fn -> "message: " <> inspect(message) end)
    end
    {:async_commit, state}
  end
end

handle_message_set/2 will be called with the batch of messages fetched from the broker. The number of messages in a batch is determined by the number of messages available and the max_bytes and min_bytes parameters of the fetch request (which can be configured in KafkaEx). In this example, because handle_message_set/2 always returns {:async_commit, new_state}, the message offsets will be automatically committed asynchronously.

committing-offsets

Committing Offsets

KafkaEx.GenConsumer manages a consumer's offsets by committing the offsets of consumed messages. KafkaEx supports two commit strategies: asynchronous and synchronous. The return value of handle_message_set/2 determines which strategy is used:

• {:sync_commit, new_state} causes synchronous offset commits.
• {:async_commit, new_state} causes asynchronous offset commits.

Note that with both of the offset commit strategies, only if the final offset in the message set is committed and this is done after the messages are consumed. If you want to commit the offset of every message consumed, use the synchronous offset commit strategy and implement calls to KafkaEx.offset_commit/2 within your consumer as appropriate.

synchronous-offset-commits

Synchronous offset commits

When handle_message_set/2 returns {:sync_commit, new_state}, the offset of the final message in the message set is committed immediately before fetching any more messages. This strategy requires a significant amount of communication with the broker and could correspondingly degrade consumer performance, but it will keep the offset commits tightly synchronized with the consumer state.

Choose the synchronous offset commit strategy if you want to favor consistency of offset commits over performance, or if you have a low rate of message arrival. The definition of a "low rate" depends on the situation, but tens of messages per second could be considered a "low rate" in most situations.

asynchronous-offset-commits

Asynchronous offset commits

When handle_message_set/2 returns {:async_commit, new_state}, KafkaEx will not commit offsets after every message set consumed. To avoid excessive network calls, the offsets are committed periodically (and when the worker terminates).

How often a KafkaEx.GenConsumer auto-commits offsets is controlled by the two configuration values :commit_interval and :commit_threshold.

• :commit_interval is the maximum time (in milliseconds) that a KafkaEx.GenConsumer will delay committing the offset for an acknowledged message.

• :commit_threshold is the maximum number of acknowledged messages that a KafkaEx.GenConsumer will allow to be uncommitted before triggering a commit.

These can be set globally in the :kafka_ex app's environment or on a per-consumer basis by passing options to start_link/5:

# In config/config.exs
config :kafka_ex,
  commit_interval: 5000,
  commit_threshold: 100

# As options to start_link/5
KafkaEx.GenConsumer.start_link(MyConsumer, "my_group", "topic", 0,
                               commit_interval: 5000,
                               commit_threshold: 100)

For low-volume topics, :commit_interval is the dominant factor for how often a KafkaEx.GenConsumer auto-commits. For high-volume topics, :commit_threshold is the dominant factor.

handler-state-and-interaction

Handler state and interaction

Use the init/2 to initialize consumer state and handle_call/3, handle_cast/2, or handle_info/2 to interact.

Example:

defmodule MyConsumer do
  use KafkaEx.GenConsumer

  defmodule State do
    defstruct messages: [], calls: 0
  end

  def init(_topic, _partition) do
    {:ok, %State{}}
  end

  def init(_topic, _partition, extra_args) do
    {:ok, %State{}}
  end

  def handle_message_set(message_set, state) do
    {:async_commit, %{state | messages: state.messages ++ message_set}}
  end

  def handle_call(:messages, _from, state) do
    {:reply, state.messages, %{state | calls: state.calls + 1}}
  end
end

{:ok, pid} = GenConsumer.start_link(MyConsumer, "consumer_group", "topic", 0)
GenConsumer.call(pid, :messages)

NOTE If you do not implement handle_call/3 or handle_cast/2, any calls to GenConsumer.call/3 or casts to GenConsumer.cast/2 will raise an error. Similarly, any messages sent to a GenConsumer will log an error if there is no corresponding handle_info/2 callback defined.

testing

Testing

A KafkaEx.GenConsumer can be unit-tested without a running Kafka broker by sending messages directly to its handle_message_set/2 function. The following recipe can be used as a starting point when testing a KafkaEx.GenConsumer:

defmodule ExampleGenConsumerTest do
  use ExUnit.Case, async: true

  alias KafkaEx.Protocol.Fetch.Message

  @topic "topic"
  @partition 0

  setup do
    {:ok, state} = ExampleGenConsumer.init(@topic, @partition)
    {:ok, %{state: state}}
  end

  test "it acks a message", %{state: state} do
    message_set = [%Message{offset: 0, value: "hello"}]
    {response, _new_state} =
      ExampleGenConsumer.handle_message_set(message_set, state)
    assert response == :async_commit
  end
end