View Source Upgrading to KafkaEx 1.0

Overview

KafkaEx 1.0 brings a cleaner API, removes legacy code, and uses Kayrock as the sole protocol implementation. This guide helps you migrate from KafkaEx 0.x.

Breaking Changes

Removed Legacy Servers

The following server implementations have been removed:

  • KafkaEx.Server0P8P0
  • KafkaEx.Server0P8P2
  • KafkaEx.Server0P9P0
  • KafkaEx.Server0P10AndLater

Kayrock is now the only implementation, providing automatic API version negotiation.

Configuration Changes

Removed options:

  • kafka_version - No longer needed; the client automatically negotiates versions

Update your config:

# Before (0.x)
config :kafka_ex,
  kafka_version: "kayrock",
  brokers: [{"localhost", 9092}]

# After (1.0)
config :kafka_ex,
  brokers: [{"localhost", 9092}]

Consumer auto_offset_reset default

The default auto_offset_reset is now :latest (matching the Kafka/Java client default), changed from :none. This only affects a consumer that has no valid committed offset and does not set auto_offset_reset explicitly. The old default behaved differently in the two cases that trigger a reset:

  • New consumer group (no committed offset): previously started silently from the earliest offset, replaying the whole topic. Now starts from the latest offset, consuming only messages produced after it joins.
  • Committed offset out of range: previously raised. Now resets to the latest offset.

:none is still available and is now consistent across both cases — it raises instead of guessing, so a missing or out-of-range offset surfaces loudly rather than silently replaying or skipping data. To replay from the beginning (the old new-group behavior), set :earliest explicitly; to keep the old strict raise-on-bad-offset behavior, set :none:

config :kafka_ex, auto_offset_reset: :earliest

Module Reorganization

Modules have been reorganized by domain:

Old ModuleNew Module
KafkaEx.GenConsumerKafkaEx.Consumer.GenConsumer
KafkaEx.ConsumerGroupKafkaEx.Consumer.ConsumerGroup
KafkaEx.New.ClientKafkaEx.Client
KafkaEx.New.KafkaExAPIKafkaEx.API
KafkaEx.New.Kafka.*KafkaEx.Messages.*

API Changes

New explicit client API:

# Before (0.x) - implicit worker
KafkaEx.produce("topic", 0, "message")
KafkaEx.fetch("topic", 0, 0)  # offset is positional

# After (1.0) - explicit client
{:ok, client} = KafkaEx.API.start_client(brokers: [{"localhost", 9092}])
{:ok, _} = KafkaEx.API.produce(client, "topic", 0, [%{value: "message"}])
{:ok, result} = KafkaEx.API.fetch(client, "topic", 0, 0)

Option key: the broker list option is :brokers (matching the config :kafka_ex, brokers: key). The pre-1.0 :uris option is still accepted as a deprecated alias. start_client/1 now merges config.exs defaults automatically, so you no longer need KafkaEx.build_worker_options/1 just to inherit your configured brokers and consumer group.

Headers API — [%Header{}] instead of [{key, value}]

The headers: option on every produce function now takes a list of %KafkaEx.Messages.Header{} structs instead of {key, value} tuples. This is a runtime breaking change — your code will compile and only fail with FunctionClauseError on the first produce. Migrate before upgrading in production.

# Before (0.x / rc.2)
KafkaEx.API.produce(client, "t", 0, [
  %{value: "v", headers: [{"trace-id", "abc"}, {"tenant", "prod"}]}
])

# After (1.0)
alias KafkaEx.Messages.Header
KafkaEx.API.produce(client, "t", 0, [
  %{value: "v", headers: [
    Header.new("trace-id", "abc"),
    Header.new("tenant", "prod")
  ]}
])

Why: the fetch path was already returning %Header{} structs. The produce side was the asymmetric outlier; a single consistent shape across produce and fetch makes round-trip code cleaner.

Broker version requirements

  • Minimum: Kafka 0.11.0+ — required for RecordBatch format, headers, and timestamps. Earlier brokers will fail at produce.
  • Tested: Kafka 2.1.0 through 3.8.x.
  • Kafka 2.3+ recommended — needed for KIP-394 two-step JoinGroup semantics with group.initial.rebalance.delay.ms. kafka_ex auto-handles the two-step dance, but broker support is required.
  • Kafka 4.0+ — partial compatibility; tracked in #497. Consumer groups may hit protocol changes.

Optional dependency matrix

Some features require additional deps in your app's mix.exs. If you configure a feature without the backing dep, you'll get an UndefinedFunctionError at runtime (not at startup).

FeatureRequired dep
Snappy compression{:snappyer, "~> 1.2"}
Zstd compression{:ezstd, "~> 1.0"}
LZ4 compression{:lz4b, "~> 0.0.13"}
MSK-IAM SASL{:jason, "~> 1.0"}, {:aws_signature, "~> 0.4"}, {:aws_credentials, "~> 1.0"}
OAuth JWT parsinguser's choice (e.g., {:joken, "~> 2.6"} — only if your token_provider needs to parse JWTs)

0.x → 1.0 API cheat-sheet

0.x1.0
KafkaEx.produce("t", 0, "m")KafkaEx.API.produce_one(client, "t", 0, "m")
KafkaEx.fetch("t", 0, offset: 0)KafkaEx.API.fetch(client, "t", 0, 0)
KafkaEx.GenConsumerKafkaEx.Consumer.GenConsumer
KafkaEx.ConsumerGroupKafkaEx.Consumer.ConsumerGroup
config :kafka_ex, kafka_version: "kayrock"(remove — no longer needed)
headers: [{"k", "v"}] on produceheaders: [Header.new("k", "v")]

OffsetCommit error handling (new in 1.0)

In earlier kafka_ex, :illegal_generation and related errors were logged and swallowed — the consumer kept running on a stale generation until the next heartbeat happened to also fail.

v1.0 classifies OffsetCommit errors across three paths, matching the reference Kafka clients (Java, librdkafka, brod, kafka-python):

  • Terminal (:fenced_instance_id, :group_authorization_failed, :topic_authorization_failed, :offset_metadata_too_large, :invalid_commit_offset_size) — consumer stops without rejoining. Under restart: :transient the supervisor does not respawn.
  • Fatal (:illegal_generation, :unknown_member_id) — GenConsumer casts {:rejoin_required, reason, stale_gen} to the group manager and self-stops. The manager resets member_id/generation_id and runs a rebalance. Duplicate casts from sibling partitions coalesce in the manager's mailbox.
  • Retryable (:rebalance_in_progress, :unstable_offset_commit, :timeout, :coordinator_not_available, …) — commit is retried with exponential backoff.

No user callback is invoked — kafka_ex v1 does not have a synchronous handle_commit_failure/3 behaviour (deferred post-1.0). Subscribe to the new telemetry event to observe failures:

:telemetry.attach(
  "my-commit-failure-observer",
  [:kafka_ex, :consumer, :commit_failed],
  fn _event, %{count: 1}, metadata, _ ->
    # metadata: %{group_id, topic, partition, offset, kind, error}
    Logger.warning("Commit failed: #{inspect(metadata)}")
  end,
  nil
)

At-least-once semantics are preserved: any uncommitted messages since the last successful commit will be redelivered after the rejoin, so your handle_message_set/2 must be idempotent (or tolerate duplicates).

GenConsumer Changes 

# Before (0.x)
defmodule MyConsumer do
  use KafkaEx.GenConsumer
  # ...
end

# After (1.0)
defmodule MyConsumer do
  use KafkaEx.Consumer.GenConsumer
  # ...
end

ConsumerGroup Changes 

# Before (0.x)
KafkaEx.ConsumerGroup.start_link(
MyConsumer, "my-group", ["topic"],
  # ...
)

# After (1.0)
KafkaEx.Consumer.ConsumerGroup.start_link(
  MyConsumer, "my-group", ["topic"],
  # ...
)

KafkaEx.API.start_client() is now unnamed by default

Pre-v1.0 patch — and the v1.0 release prior to this fix — start_client() with no :name registered the resulting GenServer as KafkaEx.Client. The :name option was documented but silently ignored.

After this change, start_client() (no :name) returns {:ok, pid} and does NOT register globally. To get the old behavior, pass :name explicitly:

# Old implicit behavior:
{:ok, _} = KafkaEx.API.start_client()
KafkaEx.API.metadata(KafkaEx.Client)  # used to work, no longer registered

# New explicit behavior:
{:ok, _} = KafkaEx.API.start_client(name: KafkaEx.Client)
KafkaEx.API.metadata(KafkaEx.Client)  # works

The :name option now also accepts {:global, term} and {:via, mod, term} shapes for distributed / registry-based registration.

Note: most users bind the returned pid ({:ok, client} = start_client(...)) and pass it to subsequent API calls. Those callers are unaffected.

Fetch timeouts now derive from :max_wait_time

Previously KafkaEx.API.fetch/5's GenServer.call timeout (5s default) and the per-broker Socket.recv timeout (1-3s from sync_timeout app config) were not aligned with the :max_wait_time option (default 10s). At logend the broker would hold the long-poll for max_wait_time ms; both upper timeouts fired first, causing socket close + retry loops until the request was abandoned. Symptom: streams built on KafkaEx.Consumer.Stream or direct KafkaEx.API.fetch/5 calls hanging or returning unexpected errors when the consumer caught up to logend.

KafkaEx.API.fetch/5 now derives both timeouts automatically:

  • network_timeout = max_wait_time + 5_000 (passed via opts to the Client GenServer, used as the Socket.recv timeout).
  • call_timeout = network_timeout × 3 + 5_000 (used as the GenServer.call timeout). The × 3 multiplier covers the Client's retry budget (@retry_count = 3) so the call does not exit while retries are still in flight, which would leak a dead-mailbox reply.

With the defaults this is max_wait_time = 10s, network_timeout = 15s, call_timeout = 50s. The high call_timeout is a worst-case ceiling, not a typical wait — a healthy fetch returns in network_timeout or less; only the retry path approaches call_timeout. If you need a tighter ceiling, pass a smaller :max_wait_time (or pass an explicit :network_timeout opt).

If you bumped :kafka_ex, :sync_timeout upward as a workaround, you can revert to the default. The sync_timeout config still governs non-fetch requests (metadata, find_coordinator, etc.) where it works correctly.

If you want fast logend halt (e.g. for short-lived streams), pass max_wait_time: 1_000 (or less) explicitly. The derived timeouts will follow.

KafkaEx.Consumer.Stream halts on fetch errors

Paired with the timeout fix: when a fetch underlying a stream returns {:error, _}, KafkaEx.Consumer.Stream now terminates cleanly via {:halt, offset} and emits a Logger.warning:

Stream halting after fetch error on <topic>/<partition> at offset <offset>: <error_code>

Previously, with no_wait_at_logend: false (default) the stream silently looped on a wildcard fallback in stream_control/3, never yielding events and never surfacing the error. With no_wait_at_logend: true the stream halted via the same wildcard but emitted no log.

Operators can grep "Stream halting after fetch error" to detect halts, or attach to [:kafka_ex, :fetch, :stop] telemetry with metadata.result == :error to alert on the underlying fetch failure (which fires regardless of stream consumption).

A related crash class is also fixed: auto_commit: true plus a fetch error response previously raised KeyError on fetch_response.message_set. need_commit?/2 now returns false for any error response, so no commit is attempted and the stream halts cleanly.

This halt applies to transient errors too — fetch timeouts, connection drops, and leader changes reach the stream only after the client's internal retries (3 attempts, with a metadata refresh on leadership errors) are exhausted. The stream itself does not retry; it halts and leaves recovery to you. (Reference clients such as the Java consumer, librdkafka, and brod retry transient errors inside the consumer loop; KafkaEx's lazy Stream delegates that to a supervisor instead. Richer in-stream recovery is tracked for a future release.)

Migration: if your application relied on the silent-spin behavior to "wait through" transient errors, wrap your stream construction in a supervisor or Task restart loop that re-enters the stream after the warning fires. Track the offset of the last message you successfully process and re-enter from the next one:

%KafkaEx.Consumer.Stream{client: client, topic: topic, partition: p,
  offset: last_processed_offset + 1, ...}

Re-entering from the original stream struct restarts from its initial :offset and reprocesses everything consumed so far (safe under at-least-once if your handler is idempotent, but usually not what you want).

Deprecations

The following functions and modules are deprecated in v1.0 and scheduled for removal in v2.0. They continue to work in the entire 1.x series — plan migration at your convenience.

DeprecatedReplacementNotes
KafkaEx.Config.consumer_group/0KafkaEx.Config.default_consumer_group/0Function-for-function swap.
KafkaEx.Client.State.max_supported_api_version/3KafkaEx.Client.State.max_supported_api_version/2Drop the default arg and match on {:ok, vsn} / {:error, :api_not_supported_by_broker}.
KafkaEx.Producer.Partitioner.LegacyKafkaEx.Producer.Partitioner.DefaultSee KafkaEx.Producer.Partitioner moduledoc.

Each of these emits an Elixir compile-time @deprecated warning — mix compile --warnings-as-errors will flag the first call site.

Migration Checklist

  • [ ] Remove kafka_version from config
  • [ ] Update KafkaEx.GenConsumer to KafkaEx.Consumer.GenConsumer (required - code will not compile)
  • [ ] Update KafkaEx.ConsumerGroup to KafkaEx.Consumer.ConsumerGroup (required - code will not compile)
  • [ ] Update code to use KafkaEx.API functions (optional but recommended)
  • [ ] Update any references to KafkaEx.New.* modules
  • [ ] If you depend on specific protocol versions, add api_versions to config (see API Version Resolution below)
  • [ ] Run tests and fix deprecation warnings
  • [ ] Verify with your Kafka cluster

Important: Old module names (KafkaEx.GenConsumer, KafkaEx.ConsumerGroup, etc.) are not aliased. Code using old module names will fail to compile immediately. All references must be updated.

New Features in 1.0

Explicit Client API

The new KafkaEx.API module provides explicit, client-based functions:

{:ok, client} = KafkaEx.API.start_client(brokers: [{"localhost", 9092}])

# Produce
{:ok, metadata} = KafkaEx.API.produce_one(client, "topic", 0, "value")

# Fetch
{:ok, result} = KafkaEx.API.fetch(client, "topic", 0, 0)

# Offsets
{:ok, offset} = KafkaEx.API.latest_offset(client, "topic", 0)
{:ok, _} = KafkaEx.API.commit_offset(client, "group", "topic", [%{partition_num: 0, offset: offset}])

# Topic management
{:ok, _} = KafkaEx.API.create_topic(client, "new-topic", num_partitions: 3)

API Version Resolution

The client now uses the highest protocol version supported by both the broker and the protocol library by default. Previous versions used conservative hardcoded defaults (e.g., fetch v3, produce v3) even when the broker supported higher versions.

If you need to pin specific API versions — for example, to match previous behavior or work around broker-specific issues — use the new api_versions application config:

config :kafka_ex,
  api_versions: %{
    fetch: 3,
    produce: 3,
    metadata: 1
  }

Version selection follows this priority order:

  1. Per-request :api_version option (highest priority)
  2. Application config api_versions map
  3. Broker-negotiated max (default)

The GenConsumer / ConsumerGroup :api_versions supervisor option continues to work for per-consumer-group overrides. Application config is no longer read by GenConsumer directly — it is handled centrally by the client's request builder.

latest_offset/4 and earliest_offset/4 no longer force list_offsets v1. They use the standard version resolution like all other API calls.

Telemetry & Observability

Built-in telemetry support for monitoring connections, requests, and consumer operations:

:telemetry.attach(
  "kafka-handler",
  [:kafka_ex, :request, :stop],
  &MyApp.handle_event/4,
  nil
)

See README.md for complete event reference and setup examples.

Compression Support

Support for multiple compression formats on a per-request basis:

# Gzip compression (built-in)
{:ok, _} = KafkaEx.API.produce(client, "topic", 0, messages, compression: :gzip)

# Supported: :gzip, :snappy, :lz4, :zstd

See README.md for details on all compression formats.

SASL Authentication

Full SASL support including PLAIN, SCRAM-SHA-256/512, OAUTHBEARER, and AWS MSK IAM:

# SCRAM example
config :kafka_ex,
  brokers: [{"localhost", 9292}],
  use_ssl: true,
  sasl: %{
    mechanism: :scram,
    username: "user",
    password: "pass",
    mechanism_opts: %{algo: :sha256}
  }

See AUTH.md for complete configuration examples for all authentication mechanisms.

Getting Help