Hi everyone!

Some time ago, I released the first version of Klife (Klife - A Kafka client with performance gains over 10x), a Kafka client written from scratch in Elixir. At that time, it only supported producing messages. Since then, I’ve been working on implementing consumer group support, and as it nears completion, I’d love to gather feedback from the community — especially around the design of the API and any Kafka-related pain points you’ve experienced, even if they’re not directly tied to Klife’s interface.

Why This Post?

This post is a starting point for collaboration — a space to discuss Kafka consumer pain points and how Klife might help. I’m sharing the proposed API and internals to collect early feedback, but also to invite anyone who’s wrestled with Kafka on the BEAM to weigh in — whether it’s design suggestions, performance headaches or any small detail that’s made Kafka harder than it should be. No detail is too small.

Basic Usage Example

Here’s a simple example of how a consumer group is defined in Klife:

defmodule MyConsumerGroup do
  use Klife.Consumer.ConsumerGroup,
    client: MyClient,
    group_name: "my_group_name",
    topics: [
      [name: "my_consumer_topic"],
      [name: "my_consumer_topic_2"]
    ]

  @impl true
  def handle_record_batch(topic, partition, record_lists) do
    # Do some processing here!
  end
end

You define a module that implements the consumer group behaviour and pass configuration either via use (compile-time) or at start_link (runtime) depending on your needs. Then start it on your supervision tree.

Consumer Group Behaviour

@type action ::
        :commit | {:skip, String.t()} | {:move_to, String.t()} | :retry

@type callback_opts :: [
        {:handler_cooldown_ms, non_neg_integer()}
      ]

@callback handle_record_batch(topic :: String.t(), partition :: integer, list(Klife.Record.t())) ::
            action
            | {action, callback_opts}
            | list({action, Klife.Record.t()})
            | {list({action, Klife.Record.t()}), callback_opts}

@callback handle_consumer_start(topic :: String.t(), partition :: integer) :: :ok
@callback handle_consumer_stop(topic :: String.t(), partition :: integer, reason :: term) :: :ok

@optional_callbacks [handle_consumer_start: 2, handle_consumer_stop: 3]

The main callback is handle_record_batch/3, where the actual record processing happens. The records are guaranteed to be ordered, and the return value tells Klife what to do with each one:

• :commit - everything went fine, commit the record

• {:skip, reason} - commit the record but note the reason for skipping it on commit metadata

• {:move_to, topic} - commit the record and send it to another topic (for retries or DLQs), also note it on commit metadata

• :retry - do NOT commit; requeue the record on the internal queue for another processing cycle with bumped attempt count

Response example:

[
  {:commit, rec1},
  {{:skip, "validation failed"}, rec2},
  {{:move_to, "my_dlq_topic"}, rec3},
  {:commit, rec4},
  {:retry, rec5}
]

If you return a single action (:commit, {:skip, reason}, etc), it will be applied to all records in the batch (i.e., it’s shorthand for the full list).

There are some edge cases to handle, such as:

• What should happen if the user returns a list shorter than the number of records?

• What if the list contains an invalid order (e.g., retrying a record before committing one with a higher offset)?

The current plan is to raise on these cases to avoid inconsistent consumer group state.

You can also return callback_opts to control things like cooldown timing (more on that below).

Consumer Group Config

A consumer group is responsible for maintaining heartbeat, reacting to rebalances, and managing consumer lifecycles. Here’s a summary of its configuration:

[
  client: [
    type: :atom,
    required: true,
    doc: "The name of the klife client to be used by the consumer group"
  ],
  topics: [
    type: {:list, {:keyword_list, TopicConfig.get_opts()}},
    required: true,
    doc: "List of topic configurations that will be handled by the consumer group"
  ],
  group_name: [
    type: :string,
    required: true,
    doc: "Name of the consumer group"
  ],
  instance_id: [
    type: :string,
    doc: "Value to identify the consumer across restarts (static membership). See KIP-345"
  ],
  rebalance_timeout_ms: [
    type: :non_neg_integer,
    default: 30_000,
    doc:
      "The maximum time in milliseconds that the kafka broker coordinator will wait on the member to revoke it's partitions"
  ],
  fetcher_name: [
    type: :atom,
    doc:
      "Fetcher name to be used by the consumers of the group. Defaults to client's default fetcher"
  ],
  committers_count: [
    type: :pos_integer,
    default: 1,
    doc: "How many committer processes will be started for the consumer group"
  ],
  isolation_level: [
    type: {:in, [:read_committed, :read_uncommitted]},
    default: :read_committed,
    doc:
      "Define if the consumers of the consumer group will receive uncommitted transactional records"
  ]
]

A few Klife-specific highlights:

• :fetcher_name - lets you group fetches to brokers under a named fetcher, enabling custom batching behavior. Similar to the current :producer_name options on the producer feature.

• :committers_count - since commit requests can not be grouped across groups, each consumer group has it’s specific commiter process which may be a bottleneck when consuming many partitions, with this option you can share the load among more committer processes

Per-Topic Consumer Options

Each consumer group manages a set of consumers—one per assigned partition. Each consumer runs its own processing loop, backed by an internal queue, and performs asynchronous fetch and commit operations to maximize throughput.

While some configuration is inherited from the consumer group, most settings are defined in the TopicConfig, which is passed via the topics option in the consumer group. These options include:

[
  name: [
    type: :string,
    required: true,
    doc: "Name of the topic the consumer group will subscribe to"
  ],
  fetcher_name: [
    type: {:or, [:atom, :string]},
    doc:
      "Fetcher name to be used by the consumers of this topic. Overrides the one defined on the consumer group."
  ],
  isolation_level: [
    type: {:in, [:read_committed, :read_uncommitted]},
    doc: "May override the isolation level defined on the consumer group"
  ],
  offset_reset_policy: [
    type: {:in, [:latest, :earliest, :error]},
    default: :latest,
    doc:
      "Define from which offset the consumer will start processing records when no previous committed offset is found."
  ],
  fetch_max_bytes: [
    type: :non_neg_integer,
    default: 50_000,
    doc:
      "The maximum amount of bytes to fetch in a single request. Must be lower than fetcher config `max_bytes_per_request`"
  ],
  fetch_interval_ms: [
    type: :non_neg_integer,
    default: 5000,
    doc: """
    Time in milliseconds that the consumer will wait before trying to fetch new data from the broker after it runs out of records to process.

    The consumer always tries to optimize fetch requests wait times by issuing requests before it's internal queue is empty. Therefore
    this option is only used for the wait time after a fetch request returns empty.

    TODO: Add backoff description
    """
  ],
  handler_cooldown_ms: [
    type: :non_neg_integer,
    default: 0,
    doc: """
    Time in milliseconds that the consumer will wait before handling new records. Can be overrided for one cycle by the handler return value.
    """
  ],
  handler_max_commits_in_flight: [
    type: :non_neg_integer,
    default: 0,
    doc: """
    Controls how many commit messages can be waiting for confirmation before the consumer stops processing new records.

    When this limit is reached, processing pauses until confirmations are received. Set to 0 to process records one batch at a time - each batch must be fully confirmed before starting the next.
    """
  ],
  handler_max_batch_size: [
    type: :pos_integer,
    default: 10,
    doc:
      "The maximum amount of records that will be delivered to the handler in each processing cycle."
  ]
]

Notable options:

• :fetch_max_bytes - Controls the size of each fetch request (not the full queue size). Actual memory use may exceed this due to async fetch prefetching.

• :fetch_interval_ms - This setting only applies when a fetch request returns no records. On busy topics, the consumer fetches on demand as soon as the internal queue drops below a threshold. But if a fetch returns empty, the consumer enters a progressive linear backoff, gradually increasing the wait time until it reaches :fetch_interval_ms, after which it waits that full interval between retries until new data becomes available.

• :handler_cooldown_ms - Adds a post-commit cooldown between batches, helping throttle consumption without blocking useful work. Aimed to address issues like the ones reported on the original post when .

• :handler_max_commits_in_flight - Allows processing new batches while waiting for previous commits to complete. A performance vs consistency tradeoff — useful when strict ordering isn’t needed. I’m also planning to add an ETS-based temporary offset store (optionally replicated cluster-wide) to reduce the risk of duplicate processing on crashes.

Caveats

• KIP-848 Only: The current implementation is built around the new rebalance protocol introduced in KIP-848, which became general available on Kafka 4.0. This gives us access to modern consumer features and address ont of the biggest pain points afaik (costly rebalances), but may limit compatibility with older clusters.

• Still a Work in Progress: The commit logic is still being finalized, but partition assignment, rebalancing, and record handling are already working well. You can check out the current implementation here: GitHub - oliveigah/klife: Kafka client for elixir · GitHub

Wrap-up

Thank you for reading this far!

Again, if you’ve struggled with Kafka in Elixir before, I’d love to hear from you — whether it’s feedback on the proposed interface, missing features in existing clients, or design tradeoffs you’d like to see better addressed. Even non-technical frustrations are valuable at this stage!

Let’s use this thread as an open forum for discussing Elixir + Kafka. Your insights will directly help shape Klife’s future direction.

Thanks in advance!

And people say programmers can’t estimate!

I’m very glad to share that the first official release of Klife’s consumer features is out. It supports Kafka >= 4.0 and already includes some of the newer capabilities from KIP-848, which should bring meaningful improvements in performance and consumer stability.

If this is useful for your use case, give it a try! Feedback is very welcome.

Thanks!

The issues usually aren’t performance but feature support

You are absolutelly right about performance not being the main issue for now!

As for features, most of the key gaps are on the consumer side, and I’m still evaluating the best approach there. I believe we can leverage BEAM’s distribution model to address certain issues uniquely—particularly rebalancing.

I have a few ideas to explore further before committing to a specific path, which may take some time. I’m estimating around six months to solidify the approach and potentially another year and a half to reach a releasable state, depending on my time availability.

There is some big changes comming for kafka such as:

• KIP-848: The Next Generation of the Consumer Rebalance Protocol - Apache Kafka - Apache Software Foundation
• KIP-932: Queues for Kafka - Apache Kafka - Apache Software Foundation

I plan to incorporate support for these from the start, leveraging the clean-slate approach I’m taking with Klife.

In short, I agree that feature support is the main pain point. For the consumer side, I can only speak to plans at this stage since it’s still in development.

For the current producer, though, I believe it’s in strong shape, with support for batch production, synchronous and asynchronous modes, idempotency with EOS, transactions, custom partitioning, and extensive performance optimizations. If there’s any producer feature you find missing, please let me know so we can look into it!

I’d love for us to have a great and robust kafka client, but at the same time I also believe that it’s too big for any one person to tackle.

I think it may be true, because Kafka is indeed a constantly evolving platform that relies on clients to take on a lot of responsibilities.

This was the main reasons I decided to build this project entirely from scratch, starting with a full protocol rewrite. To keep up with Kafka’s rapid development pace sustainably, especially with limited resources (time and money), it’s essential to have a complete understanding of the stack, end to end.

One project that’s been a big source of inspiration is Franz-go from Golang. It’s an impressive Kafka client that quickly integrates a wide range of KIPs, sometimes even outpacing the official Java client—all while being 99.9% maintained by a single developer.

I’m not sure if I’ll fully achieve this goal, but it’s definitely worth a shot—and the knowledge gained along the way has already been rewarding!