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
• Zero-copy message handling

Version Negotiation

On connection, krafka automatically fetches the broker's supported API versions and stores them. This enables dynamic version negotiation for optimal compatibility and feature usage.

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

ApiVersions is the one API whose version cannot be negotiated from a previous ApiVersions response — it is the negotiation. krafka therefore sends versions::API_VERSIONS_MAX (v4 by default) and, if the broker answers UNSUPPORTED_VERSION, retries at the ceiling that rejection advertises. The protocol mandates that a rejection is encoded with the v0 response layout precisely so a client that guessed too high can still parse the reply, so the fallback costs exactly one extra round trip and never fails the handshake.

The default ceiling is the highest version a released Kafka supports, not the highest krafka can encode: sending v5 (KIP-1242, unreleased) to a Kafka 4.x broker would cost a rejected round trip on every connection. v5 is available behind the unstable-protocol feature for testing against unreleased builds.

Negotiating v3+ rather than pinning v0 is what puts two things on the wire:

• 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; read them with BrokerConnection::broker_features(), which exposes finalized_level(name) and supported_range(name) so callers can gate optional behaviour on a cluster-wide feature level without a second round trip.

Using Version Negotiation

use krafka::protocol::ApiKey;

// negotiate_api_version(api_key, max, min) clamps to client MIN..MAX and broker range.
let fetch_version = conn
    .negotiate_api_version(ApiKey::Fetch, 12, 4)
    .await
    .expect("broker does not support any usable Fetch version");
println!("Using Fetch v{}", fetch_version);

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:

Missing any one leg silently degrades the guarantee rather than breaking visibly — the client keeps working and simply stops noticing truncation. The commit leg is the easiest to overlook, because its absence is invisible until a consumer restarts.

A fenced ListOffsets forces a metadata refresh before the retry, so the epoch check converges instead of failing identically forever.

-1 remains the correct value where the client genuinely has no epoch: a position that came from a seek() or an offset reset, or an offset set administratively through AdminClient::alter_consumer_group_offsets. Inventing one there would defeat the check it feeds.

Minimum Broker Version

krafka requires Apache Kafka 3.9 or later. The MIN constants for all APIs are set so that pre-3.9 protocol features (e.g., Metadata v0, Produce v0-v2, Fetch v0-v3) are no longer supported. Connecting to an older broker will fail version negotiation for most APIs.

Client Supported Versions

Every API has a MIN and MAX constant in krafka::protocol::versions. The client only encodes/decodes versions within [MIN, MAX]; versions outside this range are rejected with a protocol error.

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

² Requires the telemetry feature flag.

The share-group offset administration APIs (keys 90–92) are not behind unstable-protocol: they are ordinary AdminClient operations and are compiled unconditionally. Only the ShareConsumer itself is gated.

StreamsGroupDescribe (key 89) is likewise ungated — it is an ordinary AdminClient operation. Its sibling StreamsGroupHeartbeat (key 88) is deliberately not implemented: see below.

KIP-1071: why only the describe half

KIP-1071 adds two APIs, and krafka implements exactly one of them.

StreamsGroupHeartbeat (key 88) is a Streams runtime API. Its request carries the application topology — subtopologies, repartition topics, changelog topics — and the coordinator assigns tasks from it. Sending it requires being a Streams runtime. A client that fabricated a topology would not merely be wrong locally: the topology is group-wide state, so it would corrupt what every real member of that group is assigned from. krafka has no Streams layer, so it does not send this.

StreamsGroupDescribe (key 89) is purely observational and is what an operator actually needs — see Admin Client → Streams Groups.

How this table stays honest

Two mechanisms, neither of which is "someone remembered".

Within the crate, the rows above are generated from the same api_versions! macro tokens that initialise the *_MIN / *_MAX constants, so the published table cannot understate or overstate what the client negotiates.

Against Kafka, just protocol-parity diffs the table against Apache Kafka's own message schemas and fails CI on any of five conditions:

The check reads a vendored snapshot (xtask/kafka_protocol_snapshot.json), so it needs no network and cannot flake. Track a newer Kafka release deliberately:

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.

Why some ceilings stop short

A schema marked latestVersionUnstable: true is not advertised by a released broker unless it was started with unstable.api.versions.enable=true, so implementing it buys nothing and costs a rejected round trip on every connection.

In Kafka 4.3 exactly one API carries that flag: InitProducerId v6 (KIP-939 two-phase commit), which is why it sits behind unstable-protocol. ListOffsets v11, AddPartitionsToTxn v5, EndTxn v5 and DescribeQuorum v2 set it explicitly to false — they are stable, and krafka negotiates all of them.

Version Constants

Client-supported versions are defined in krafka::protocol::versions:

use krafka::protocol::versions;

// Each API has both MIN and MAX constants
let min_fetch = versions::FETCH_MIN;        // 4  (Kafka 3.9+ baseline)
let max_fetch = versions::FETCH_MAX;        // 18 (v18 KIP-1166 high-watermark)
let min_produce = versions::PRODUCE_MIN;    // 3  (v3+ transactions)
let max_produce = versions::PRODUCE_MAX;    // 13 (v13 topic UUIDs, KIP-516)
let max_metadata = versions::METADATA_MAX;  // 13 (v13 top-level error_code)

Record Batches

krafka uses Kafka's v2 record batch format with:

• Magic byte 2 (modern format)
• CRC32C checksums (validated on decode)
• Variable-length encoding for efficiency
• 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

Core request/response message types in krafka::protocol implement the VersionedEncode and VersionedDecode traits, which dispatch to the correct encode_vN/decode_vN method based on the protocol version number:

use krafka::protocol::{VersionedEncode, VersionedDecode, MetadataRequest, MetadataResponse};

let request = MetadataRequest::all_topics();
let mut buf = bytes::BytesMut::new();

// Encode for a specific protocol version — dispatches to the right encoder
request.encode_versioned(1, &mut buf)?;

// In real usage, `response_buf` would be filled with bytes read from the network.
let mut response_buf = buf.freeze();

// Decode response for a specific version
let response = MetadataResponse::decode_versioned(1, &mut response_buf)?;

Unsupported version numbers (including negative values) return a descriptive KrafkaError::protocol error.

Creating Records

use krafka::protocol::{RecordBatchBuilder, Compression};

let batch = RecordBatchBuilder::new()
    .compression(Compression::Snappy)
    .add_record(Some(b"key"), Some(b"value"), vec![])
    .add_record(None, Some(b"value-only"), vec![])
    .build()?;

Compression Support

Note: Decompression output is capped at 128 MiB by default to protect against compression bombs. This limit is configurable via ConsumerConfig::max_decompressed_size(). Compressed payloads that expand beyond the limit will return a KrafkaError::compression error.

Protocol Safety

krafka protects against malicious or corrupted broker responses:

• Decode array bounds: Every array-length field decoded from the wire is validated against MAX_DECODE_ARRAY_LEN (100,000), typically via check_decode_array_len() and in some specialized decode paths (e.g., KafkaArray::decode, record batch counts) via equivalent local checks. These checks reject negative counts and oversized counts across all 63+ protocol-message decode sites, KafkaArray decode paths, and record batch/header counts. The validation runs before any Vec::with_capacity() allocation, preventing both OOM and runaway decode loops.
• Decompression limits: Decompressed record data is limited to 128 MiB (configurable) via streaming .take() limits and post-decompression size checks
• Record headers: Record headers are preserved during batch building — no silent data loss
• Encode validation: The TryEncode trait provides fallible encoding for protocol primitives (KafkaString, KafkaBytes, KafkaArray<T> where T: TryEncode, TaggedFields), returning errors instead of panicking on oversized data. ProducerRecord::validate() checks wire-format limits at the API boundary before encoding
• Fuzz testing: The fuzz/ directory provides cargo-fuzz targets for KafkaArray decode, RecordBatch decode, and response message decode across multiple API versions. 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)     |
+---------------+

Zero-Copy Design

krafka uses bytes::Bytes throughout for zero-copy buffer management:

• Incoming data is parsed without copying
• Record payloads share underlying buffers
• Memory is released when last reference drops

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-matrix "4.2.0 4.3.0"    # a subset
KAFKA_VERSION=4.3.0 just integration     # a single version

Redpanda

Redpanda works out of the box — no configuration, no feature flag:

• Version negotiation lands inside Redpanda's advertised ranges for every API krafka's clients need (krafka's floors — Produce v3, Fetch v4, Metadata v1 — sit well below what current Redpanda advertises).
• 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 at init_transactions() and uses the classic explicit-AddPartitionsToTxn protocol when the feature is absent — the same fallback the Java 4.x client performs.
• KIP-848 consumer groups are implemented by current Redpanda, so the next-generation group protocol works there too; the classic protocol is the fallback either way.
• What does not apply: APIs Redpanda does not implement — KIP-932 share groups (ShareConsumer), log-dir administration, DescribeQuorum — fail fast with a clear UnknownApiVersion error rather than degrading silently, exactly as they do against an older Apache Kafka.

A dedicated smoke suite pins this against a real Redpanda container, including the TV1 fallback and read_committed visibility:

just integration-redpanda
REDPANDA_VERSION=v25.1.1 just integration-redpanda   # pin a tag

Next Steps

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