Kafka
Apache Kafka uses a binary request-response protocol over TCP. Every request carries a header identifying the API being called and its version, followed by an API-specific body.
The protocol has no global version. Each API is versioned independently, and a client negotiates per API at connection time by issuing an ApiVersions request and using the highest version both sides support. DataStream negotiates in this way, so the exact wire layout depends on the broker.
Request Header
| Field | Type | Description |
|---|---|---|
length | int32 | Size of the request, excluding this field |
api_key | int16 | Identifies the API being called |
api_version | int16 | Version of that API |
correlation_id | int32 | Echoed in the response, to match the two |
client_id | nullable_string | Optional client identifier |
tagged_fields | tagged_fields | Header v2 only |
Three header versions exist. v0 ends at correlation_id, v1 adds client_id, and v2 adds the trailing tagged-fields buffer. Every API supporting flexible versions uses v2, so a capture decoded against v1 will mis-parse most modern requests.
Response Header
| Field | Type | Description |
|---|---|---|
length | int32 | Size of the response, excluding this field |
correlation_id | int32 | Matches the request |
tagged_fields | tagged_fields | Header v1 only |
body | varies | Depends on the API. Most bodies open with throttle_time_ms |
API Keys
The APIs DataStream exercises, producing or consuming:
| API Key | Name | Purpose |
|---|---|---|
0 | Produce | Send records to a topic |
1 | Fetch | Retrieve records from a topic |
2 | ListOffsets | Get the earliest or latest offset for a partition |
3 | Metadata | Get topic and partition information |
8 | OffsetCommit | Commit consumer group offsets |
9 | OffsetFetch | Retrieve committed offsets |
10 | FindCoordinator | Locate the group coordinator |
11 | JoinGroup | Join a consumer group |
12 | Heartbeat | Maintain group membership |
13 | LeaveGroup | Leave a consumer group |
14 | SyncGroup | Receive the partition assignment |
17 | SaslHandshake | Select a SASL mechanism |
18 | ApiVersions | Discover the versions the broker supports |
36 | SaslAuthenticate | Exchange SASL credentials |
Primitive Types
| Type | Description |
|---|---|
int8, int16, int32, int64 | Signed integers, big-endian |
boolean | A single byte, 0 or 1 |
float64 | IEEE 754 double, big-endian |
uuid | 16 raw bytes |
string | int16 length prefix, then UTF-8 |
nullable_string | As string, with -1 meaning null |
bytes | int32 length prefix, then raw bytes |
array | int32 element count, then the elements |
varint, varlong | Variable-length integer, zigzag encoded |
unsigned_varint | Variable-length integer, not zigzag encoded |
compact_string | unsigned_varint length prefix, then UTF-8 |
compact_array | unsigned_varint element count, then the elements |
records | A record batch |
tagged_fields | unsigned_varint count, then tag, length and data triples |
The compact forms and tagged_fields appear only on flexible-version APIs. Note that varint is zigzag encoded and unsigned_varint is not, despite the similar name.
Authentication
SASL is negotiated with SaslHandshake followed by SaslAuthenticate. The following mechanisms are supported:
| Mechanism | Description |
|---|---|
plain | Username and password in the clear. Use with TLS |
scram-sha-256 | Salted challenge-response |
scram-sha-512 | Salted challenge-response with a stronger digest |
gssapi | Kerberos |
OAUTHBEARER and AWS_MSK_IAM are not supported.
Consumer Groups
Consumption uses consumer groups, with offsets committed explicitly rather than automatically. The partition assignment strategy is one of cooperativesticky, range, roundrobin or sticky. The names carry no hyphens: matching is case-insensitive but exact, and any other value — including the hyphenated spellings used elsewhere in the Kafka ecosystem — silently falls back to roundrobin rather than being rejected.