Protocol Support

Supported Kafka APIs and versions, negotiation, and how parity with Apache Kafka is enforced in CI.

Overview

krafka implements the Kafka wire protocol with support for:

• Automatic API version negotiation
• Multiple protocol versions per API
• All standard compression codecs, decoded in pure Rust
• Decoded records that share the fetched buffer

Version Negotiation

On connection, krafka fetches the broker's supported API versions and picks, per request, the highest version both sides support.

How It Works

1. Client connects to broker
2. Client sends ApiVersions request
3. Broker responds with supported API version ranges
4. Client stores version ranges for future requests
5. Each request negotiates the best version within the client's [MIN, MAX] range

Bootstrapping ApiVersions itself

krafka sends ApiVersions at its own ceiling (v4; v5 with the unstable-protocol feature). A broker that answers UNSUPPORTED_VERSION encodes the rejection in the v0 layout with its own range, and krafka retries once at that ceiling.

v3 and later carry:

• KIP-511 — ClientSoftwareName / ClientSoftwareVersion, which is how a broker's client.software.name / client.software.version metrics identify krafka. These fields do not exist below v3.
• KIP-584 — SupportedFeatures and FinalizedFeatures, carried in v3+ tagged fields. krafka caches them per connection and reads them itself — the transactional producer picks its transaction protocol from the finalized transaction.version. Applications read them with AdminClient::describe_features.

Negotiation is internal: the connection layer is not part of the public API.

Leader epochs end to end (KIP-320)

KIP-320 only detects log truncation if the leader epoch travels with the position everywhere it goes. krafka sends it on all three legs:

A fenced ListOffsets refreshes metadata before the retry.

The epoch is -1 where the client has none: a position from a seek() or an offset reset, or an offset set through AdminClient::alter_consumer_group_offsets.

Minimum Broker Version

krafka requires Apache Kafka 3.9 or later. Each API's minimum version is one a Kafka 3.9 broker accepts (for example Metadata v1, Produce v3, Fetch v4), so an older broker fails version negotiation for most APIs.

Client Supported Versions

Every API has a minimum and a maximum supported version. The client only encodes/decodes versions within [MIN, MAX]; versions outside this range are rejected with a protocol error.

OffsetFetch and OffsetCommit v10 require nonzero topic IDs for explicit topics; invalid IDs fail encoding before bytes are written. Null OffsetFetch topics request all offsets. Admin group-offset calls use name-based versions through v9.

¹ Requires the unstable-protocol feature flag — the version Kafka marks latestVersionUnstable. Where a max is shown in parentheses, that is the feature-gated ceiling.

The share-group APIs are not behind unstable-protocol — Kafka marks none of them unstable, and krafka negotiates them on every build.

StreamsGroupDescribe (key 89) is likewise ungated; see Admin Client → Streams groups. StreamsGroupHeartbeat (key 88) is not implemented: its request carries the application topology, which only a Streams runtime can supply.

How the table is checked

Two checks, both in CI.

Against the crate, just site-check compares every row above with the api_versions! table in src/protocol/mod.rs, which defines the MIN and MAX constants the client negotiates with: an implemented API missing from this page, a different minimum, a ceiling the page does not mention, or a row for an API the crate does not implement fails it.

Against Kafka, just protocol-parity (xtask/protocol_parity.py) diffs the api_versions! table against Apache Kafka's own message schemas and fails on any of six conditions:

It also fails when Kafka has an API with neither a row in api_versions! nor an entry in DELIBERATE_GAPS, and when Kafka has removed an API the client still negotiates.

The check reads a vendored snapshot (xtask/kafka_protocol_snapshot.json) and needs no network. To track a newer Kafka release:

just refresh-protocol-snapshot 4.3   # rewrite the snapshot; review the diff
just protocol-parity                 # see what krafka must do about it

Deliberate omissions live in a DELIBERATE_GAPS table in the script, each with a written reason — broker-internal APIs, the KIP-1071 Streams protocol, the legacy AlterConfigs that IncrementalAlterConfigs supersedes.

Unstable versions

A released broker advertises a version marked latestVersionUnstable only with unstable.api.versions.enable=true. Such versions sit behind the unstable-protocol feature: in Kafka 4.3 that is InitProducerId v6 (KIP-939 two-phase commit). ApiVersions v5 (KIP-1242), which no released Kafka has, is gated the same way.

KIP support

Every KIP this documentation names, generated from xtask/kips.toml. Each status cites the evidence that holds it — an api_versions! row, a Cargo feature or a public item — and just claims-check fails if the evidence does not resolve, or if this table is edited by hand. Tracked against

Not implemented

• SASL/GSSAPI (Kerberos) authentication (GSSAPI) — No mature pure-Rust GSSAPI implementation exists, and linking system Kerberos libraries would add a C dependency for every user.
• Schema registry client (schema-registry) — A schema registry is a separate service with its own protocol; krafka provides the serdes::Serializer and Deserializer hooks instead.
• Async runtimes other than Tokio (runtime-agnostic) — krafka is built on Tokio and does not abstract over the async runtime.
• Kafka Streams runtime (streams-runtime) — krafka is a client library with no stream-processing runtime, which is also why StreamsGroupHeartbeat is not implemented.
• Broker-, controller- and KRaft-internal APIs (broker-internal-apis) — APIs such as LeaderAndIsr, UpdateMetadata, Vote and the share-group state persister are spoken between brokers, not by clients.
• KIP-368 Allow SASL connections to periodically re-authenticate, partly — The broker-reported session lifetime is honoured by replacing a pooled connection before it expires; in-band re-authentication of a live connection is not implemented.
• KIP-525 Return topic metadata and configs in CreateTopics response, partly — CreateTopics v5+ is negotiated, but create_topics returns only per-topic success, so the partition count, replication factor and configs in the response are not surfaced.
• KIP-794 Strictly uniform sticky partitioner, partly — Keyless records stick to a partition for batch_size bytes and then switch at random, but the next partition is not weighted by per-broker queue size (partitioner.adaptive.partitioning.enable) and slow brokers are not avoided (partitioner.availability.timeout.ms).
• KIP-800 Add reason to JoinGroupRequest and LeaveGroupRequest, partly — JoinGroup v8 and LeaveGroup v5 are negotiated, but the reason field is always sent as null.
• KIP-853 KRaft controller membership changes, partly — Quorum membership can be described, but the AddRaftVoter and RemoveRaftVoter admin RPCs are not implemented.
• KIP-932 Queues for Kafka, partly — The share consumer and share-group offset administration are implemented, but ShareGroupDescribe is never sent, so no admin call describes or lists share groups' members.
• KIP-1071 Streams rebalance protocol, partly — Streams groups can be described, but StreamsGroupHeartbeat is not implemented because its request carries an application topology that only a Streams runtime can supply.
• KIP-1242 Detection and handling of misrouted connections, partly — ApiVersions v5 is encoded behind unstable-protocol, but the ClusterId and NodeId fields are never populated and REBOOTSTRAP_REQUIRED from ApiVersions does not trigger a rebootstrap.
• AlterConfigs is implemented below Kafka's ceiling — superseded by IncrementalAlterConfigs, which krafka uses instead; the legacy whole-config replace is not exposed
• SaslHandshake is implemented below Kafka's ceiling — pinned at v1 by the handshake path; v0 has no mechanism list
• SaslAuthenticate is implemented below Kafka's ceiling — pinned at v1: v2 only adds flexible encoding, and the pre-auth reader is deliberately version-pinned so an unauthenticated peer cannot steer it
• Not spoken by a client (broker-, controller- and KRaft-internal): LeaderAndIsr, StopReplica, UpdateMetadata, ControlledShutdown, Vote, BeginQuorumEpoch, EndQuorumEpoch, AlterPartition, Envelope, FetchSnapshot, BrokerRegistration, BrokerHeartbeat, UnregisterBroker, AllocateProducerIds, ControllerRegistration, AssignReplicasToDirs, UpdateRaftVoter, InitializeShareGroupState, ReadShareGroupState, WriteShareGroupState, DeleteShareGroupState, ReadShareGroupStateSummary

Record Batches

krafka uses Kafka's v2 record batch format with:

• Magic byte 2
• CRC32C checksums, validated on decode
• Varint-encoded record fields
• Optional compression (gzip, snappy, lz4, zstd)

Header Versioning

Every Kafka request/response is prefixed with a header whose format depends on whether the API version uses flexible encoding:

The transition version varies per API (e.g., Fetch becomes flexible at v12, Produce at v9). ApiKey::flexible_version() returns the threshold for each API, and the header is selected automatically by RequestHeader::encode() / ResponseHeader::decode().

Note: ApiVersions response always uses header v0 regardless of the API version (needed for protocol bootstrapping).

Unified Version Dispatch

Every request and response type encodes and decodes each supported version through one entry point that dispatches on the negotiated version number. Unsupported version numbers (including negative values) return a descriptive KrafkaError::Protocol error.

The protocol layer is internal: it is not part of krafka's public API and is not covered by semver.

Compression Support

Note: Decompression output is capped at 128 MiB by default to protect against compression bombs. This limit is configurable with the consumer builder's max_decompressed_size. Compressed payloads that expand beyond the limit will return a KrafkaError::compression error.

Protocol Safety

Limits on what a malicious or corrupted broker response can make the client do:

• Decode array bounds: Every array length decoded from the wire is checked against MAX_DECODE_ARRAY_LEN (100,000) before anything is allocated for it.
• Record batch bounds: A record batch has no record-count limit; it is bounded by its bytes. A declared count larger than the record bytes can hold (7 bytes per record at least) is rejected before anything is allocated for it, and decompressed output is bounded by the size limit below.
• Decompression limits: Decompressed record data is limited to 128 MiB (configurable) via streaming .take() limits and post-decompression size checks
• Encode validation: The TryEncode trait provides fallible encoding for protocol primitives (KafkaString, KafkaBytes, KafkaArray<T> where T: TryEncode, TaggedFields), returning an error on oversized data. Record::validate() checks wire-format limits at the API boundary before encoding
• Fuzz testing: The fuzz/ directory provides cargo-fuzz targets for the framing header and primitives, KafkaArray decode, RecordBatch decode, every response decoder at every version in the table above, request encode, the SCRAM exchange, the OIDC token client's HTTP response parser, and record-batch decompression under a small cap in every codec. just fuzz-coverage fails when a version in the table has no fuzz path. Every pull request runs each target for 60 seconds from committed seeds. See fuzz/README.md for usage.

Wire Protocol

Request/Response Framing

+----------------+----------------+
|  Size (4B)     |  Data (N bytes)|
+----------------+----------------+

All messages are length-prefixed with a 4-byte big-endian size field.

Request Header

+----------+----------+---------------+-------------+
| API Key  | Version  | Correlation ID| Client ID   |
| (2 bytes)| (2 bytes)| (4 bytes)     | (variable)  |
+----------+----------+---------------+-------------+

Response Header

+---------------+
| Correlation ID|
| (4 bytes)     |
+---------------+

Buffers

• The reader reserves each response frame once, at the size its length prefix declares, and reads socket data straight into it: a 16 MiB frame is one allocation of 1.01× its size.
• A record batch's header is parsed first; the consumer skips aborted transactional batches from the header, without decompressing them.
• Decoded record keys, values and header values are slices of the response buffer (uncompressed) or of the decompressed buffer. Decoding 1000 uncompressed 100-byte records makes one allocation (the record list), 0.89× the wire size.
• A record kept alive keeps the buffer it came from alive. Copy the bytes you retain long-term if the rest of the response should be freed.

Broker Compatibility

krafka negotiates every API version rather than pinning them, so it works against any broker that speaks the Kafka wire protocol at the negotiated range — Apache Kafka 3.9+, and Kafka-compatible systems whose advertised versions overlap krafka's floors.

Apache Kafka 3.9 → 4.3

Apache Kafka is the reference target: the version table above is diffed against Kafka's own message schemas in CI (just protocol-parity), and the Docker integration suite can be run against every supported minor in one command:

just integration-matrix                  # Kafka 3.9.0 → 4.3.0
just integration-sasl-matrix             # SASL suite on Kafka 3.9.0 and 4.3.1
just integration-matrix "4.2.0 4.3.0"    # a subset
KAFKA_VERSION=4.3.0 just integration     # a single version

Redpanda

Redpanda needs no configuration and no feature flag:

• Version negotiation lands inside Redpanda's advertised ranges for every API krafka's clients need.
• Transactions fall back to TV1 automatically. Redpanda does not implement server-side KIP-890 transaction version 2; krafka's TransactionalProducer probes the cluster's finalized transaction.version feature in build_transactional and uses the classic explicit-AddPartitionsToTxn protocol when the feature is absent, as the Java 4.x client does.
• APIs Redpanda does not implement (KIP-932 share groups, log-dir administration, DescribeQuorum) fail with UnknownApiVersion; see A cluster without a feature.

The Redpanda suite covers a produce/consume round trip, the admin topic lifecycle, the TV1 transaction fallback with read_committed visibility, and closing a consumer without wedging a shared connection. CI gates every merge on the Redpanda release pinned in tests/redpanda/Dockerfile; the same suite runs locally against latest:

just integration-redpanda                           # the pinned release
REDPANDA_VERSION=latest just integration-redpanda   # the current release

A cluster without a feature

A broker that lacks an API, or refuses a feature, fails the call with an error whose message names the missing feature and, where one exists, the setting that avoids it. The error keeps its kind (UnknownApiVersion when the API cannot be negotiated) and the broker's error code.

Next Steps

• Producer Guide - Sending messages
• Consumer Guide - Receiving messages
• Configuration Reference - All settings