Digging into Kafka's Source 3: The Wire Protocol — Requests and Responses Generated from JSON Schemas

What this post covers

This is the third post in the "Digging into Kafka's Source" series. In part 1 we fetched the source and built it at a fixed commit, and in part 2 we drew the big map of how clients, brokers and controllers divide the work. With this post we start the series' second group, "the path of one record". The first step is the path itself: the agreement about the bytes clients and brokers exchange over TCP.

We want to answer four questions.

• What does a producer's request look like as bytes, and how does the broker know where one request ends?
• Why are classes like ProduceRequestData missing from the repository, and where do they come from?
• What lets a client and a broker talk even when they run different Kafka versions? (ApiVersions)
• What are "flexible versions" and "tagged fields", and why were they added later?

All line numbers in this post refer to this commit. Part 1 shows how to get the same commit and build the Docker image used for the examples.

A close-up of the "client ↔ broker" arrow on the part 2 map. The yellow boxes are this post's scope, the blue box is part 4, and the grey boxes come up in other parts.

This post zooms into the single arrow drawn between region A (clients) and region B (the broker's network layer) on the part 2 map — the "between A and B" row. The key point is that the side drawing the arrow (NetworkClient) and the side receiving it (the broker's Processor) use the same generated classes. Which broker thread passes the received bytes on, and how, is the topic of part 4.

The protocol code is scattered across the clients module, and the generated code alone is over 180,000 lines. This post only follows the main path a first-time reader can keep up with; what was left out is listed by name and location under "What this post leaves out" at the end.

Terms to know first

The wire protocol is "the rules for the bytes that travel over the wire". Kafka clients and brokers don't speak HTTP; they talk over TCP using a binary protocol Kafka defines itself. Brokers talking to each other (a follower fetching from the leader for replication) and brokers talking to controllers use the same protocol.

An API key is a 16-bit number naming the kind of request. Produce is 0, Fetch is 1, Metadata is 3, ApiVersions is 18. This commit has 95 of them, from 0 to 94; four of those (LeaderAndIsr and others) were removed in Kafka 4.0 and only keep their number reserved, with no versions left ("validVersions": "none"). A response has no API key of its own; it is taken to be the same as its request's.

A version is a 16-bit number each API keeps separately. This commit knows Produce requests v3 through v13 and Fetch v4 through v18. When fields are added or removed, or their meaning changes, the version goes up. A response's version equals its request's (with one exception, an ApiVersions request in a version the broker doesn't know, covered in section 5).

A correlation ID is a 32-bit number the client attaches to every request. The broker returns the same number in its response. Since one connection can have several requests sent and waiting for responses, this number confirms which request a response belongs to (section 1).

The request header and response header are small prefixes in front of every request and response. The request header holds the API key, the version, the correlation ID and the client ID; the response header holds only the correlation ID. Headers have versions too (section 4).

Generated code: classes whose names end in …Data, like ProduceRequestData and ApiVersionsResponseData, were not written by a person. They are produced at build time from message definitions (schemas) written in JSON, so they are neither in the repository nor on GitHub. When this post quotes generated code, it copies from files generated at the reference commit and links to the original JSON schema.

varint: an integer encoding that writes small numbers in few bytes. Each byte carries 7 bits, and a set top bit means "continued in the next byte". Up to 127 takes 1 byte, up to 16,383 takes 2 (ByteUtils.writeUnsignedVarint(), from ByteUtils.java line 408). The "flexible versions" of section 6 use it to write lengths.

1. The shape of the bytes — the length goes first

Every Kafka request and response, without exception, comes in the same frame.

[4-byte length N][N bytes: header + body]

Let's start with the sender. The client turns a request into bytes for the socket in SendBuilder.buildSend().

    private static Send buildSend(
        Message header,
        short headerVersion,
        Message apiMessage,
        short apiVersion
    ) {
        ObjectSerializationCache serializationCache = new ObjectSerializationCache();

        MessageSizeAccumulator messageSize = new MessageSizeAccumulator();
        header.addSize(messageSize, serializationCache, headerVersion);
        apiMessage.addSize(messageSize, serializationCache, apiVersion);

        SendBuilder builder = new SendBuilder(messageSize.sizeExcludingZeroCopy() + 4);
        builder.writeInt(messageSize.totalSize());
        header.write(builder, serializationCache, headerVersion);
        apiMessage.write(builder, serializationCache, apiVersion);

        return builder.build();
    }

clients/src/main/java/org/apache/kafka/common/protocol/SendBuilder.java lines 208–226

It makes two passes.

1. Count the size first. It calls addSize() on the header and the body and adds up how many bytes they will take. Along the way it stores each string's UTF-8 bytes in serializationCache.
2. Allocate the buffer once and write. It writes the total length first (writeInt), then the header and the body. Strings come out of the cache filled in step 1, so they are not converted twice.

Notice that the buffer size is sizeExcludingZeroCopy(), not totalSize(). Large chunks that already sit in a ByteBuffer, like the record batches in a Produce request, are not copied into the new buffer but attached separately (the SendBuilder class comment calls these "zero-copy" fields). They count toward the length but are never copied into the buffer. What a record batch looks like inside is covered in part 6.

The receiver is NetworkReceive. Its class comment sums it up in one line: "A size delimited Receive that consists of a 4 byte network-ordered size N followed by N bytes of content".

    public long readFrom(ScatteringByteChannel channel) throws IOException {
        int read = 0;
        if (size.hasRemaining()) {
            int bytesRead = channel.read(size);
            if (bytesRead < 0)
                throw new EOFException();
            read += bytesRead;
            if (!size.hasRemaining()) {
                size.rewind();
                int receiveSize = size.getInt();
                if (receiveSize < 0)
                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + ")");
                if (maxSize != UNLIMITED && receiveSize > maxSize)
                    throw new InvalidReceiveException("Invalid receive (size = " + receiveSize + " larger than " + maxSize + ")");
                requestedBufferSize = receiveSize; // may be 0 for some payloads (SASL)
                // ...
            }
        }
        if (buffer == null && requestedBufferSize != -1) { // we know the size we want but haven't been able to allocate it yet
            buffer = memoryPool.tryAllocate(requestedBufferSize);
            // ...
        }
        if (buffer != null) {
            int bytesRead = channel.read(buffer);
            // ...

clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java lines 82–108

It reads until the 4-byte size buffer is full, then allocates a buffer of that size and reads again. The socket is non-blocking, so the data may not arrive all at once and this method is called many times; the space left in size and buffer remembers "how far we've read". complete() is true when both are full (lines 78–80).

Because the length goes first, the receiver knows where one request ends without decoding any of its content. That makes it easy to separate receiving bytes (the network thread) from interpreting them, and lets the receiver reject a request on its size alone. The broker passes socket.request.max.bytes as maxSize; the default is 100MB (SocketServerConfigs.java line 96). With HTTP/1.1, as we saw in Tomcat part 6, the server had to scan the text request line and headers character by character, looking for the blank line, to find where the headers end. That is why binary protocols use a length prefix.

Responses carry no API key — so order matters

The response header holds only the correlation ID. From the response bytes alone you can't tell which kind of request it answers or which version to decode it with. The client takes that information from the request it sent.

    private void handleCompletedReceives(List<ClientResponse> responses, long now) {
        for (NetworkReceive receive : this.selector.completedReceives()) {
            String source = receive.source();
            InFlightRequest req = inFlightRequests.completeNext(source);

            AbstractResponse response = parseResponse(receive.payload(), req.header);

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java lines 1097–1102

InFlightRequests keeps the sent requests in a deque per node. On send, add() puts the request in with addFirst() (lines 45–50); when a response arrives, completeNext() takes the oldest request with pollLast() (InFlightRequests.java lines 65–69). The response is decoded with that request's header, and AbstractResponse.parseResponse() checks that the correlation IDs match, throwing CorrelationIdMismatchException if they don't (AbstractResponse.java lines 97–108).

In other words, the client trusts that "responses on one connection arrive in the order the requests were sent". The correlation ID is the safety net that notices when that trust is broken. How the broker keeps this order (handling only one request per connection at a time with mute/unmute) is covered in part 4.

A response doesn't say which request it answers. The client pairs it with the oldest request in the deque, and the correlation ID only checks that the pair is right.

Example 1: send three requests without waiting for responses

The examples in this post were run in a Docker container (Eclipse Temurin JDK 25) with Kafka 4.5.0-SNAPSHOT built as-is from the reference commit (ab1e816411). The example folder is ex/u03-wire-protocol, run against a server with the broker and controller in one process (broker listener 9092, controller listener 9093).

The star is RawWire.java. It uses neither KafkaProducer nor NetworkClient; it builds the request bytes with nothing but the generated …Data classes and writes them to a java.net.Socket. The byte-building part reproduces SendBuilder.buildSend() above by hand.

// An example written for this post (not part of Kafka's source)
    // [4-byte length][request header][request body] — doing SendBuilder.buildSend()'s job by hand
    static byte[] frame(short apiVersion, ApiMessage body) {
        ApiKeys apiKey = ApiKeys.forId(body.apiKey());
        short headerVersion = apiKey.requestHeaderVersion(apiVersion);
        RequestHeaderData header = new RequestHeaderData()
            .setRequestApiKey(apiKey.id)
            .setRequestApiVersion(apiVersion)
            .setCorrelationId(nextCorrelationId++)
            .setClientId("u03");
        ObjectSerializationCache cache = new ObjectSerializationCache();
        // 1) count the size first (this puts the strings' UTF-8 bytes into the cache)
        int size = header.size(cache, headerVersion) + body.size(cache, apiVersion);
        ByteBuffer buf = ByteBuffer.allocate(4 + size);
        ByteBufferAccessor w = new ByteBufferAccessor(buf);
        // 2) write the length, the header, then the body
        w.writeInt(size);
        header.write(w, cache, headerVersion);
        body.write(w, cache, apiVersion);
        return buf.array();
    }

[Abridged from the example ex/u03-wire-protocol/RawWire.java]

On the receiving side, readFrame() reads the length with readInt() and then exactly that many bytes with readFully() — the same rule as NetworkReceive.readFrom().

The pipeline mode writes an ApiVersions request (what it is comes in section 5) three times in a row, and only then reads the three responses.

=== 8. pipelining: three requests, then three responses
response 1: correlationId=1
response 2: correlationId=2
response 3: correlationId=3

The responses came back in the order sent, each carrying the correlation ID it was sent with. It looks just like the HTTP/1.1 pipelining example in Tomcat part 6.

2. Reading one JSON schema

The fields of requests and responses are written in the JSON files under clients/src/main/resources/common/message/. This commit has 203 JSON files (plus one README.md in the same folder): 95 requests, 95 responses, the two headers (RequestHeader, ResponseHeader), and 11 data structures that aren't messages (ConsumerProtocolSubscription, KRaft's VotersRecord and so on).

Let's look at the smallest request, ApiVersions, the star of section 5.

  "apiKey": 18,
  "type": "request",
  "listeners": ["broker", "controller"],
  "name": "ApiVersionsRequest",
  // Versions 0 through 2 of ApiVersionsRequest are the same.
  //
  // Version 3 is the first flexible version and adds ClientSoftwareName and ClientSoftwareVersion.
  //
  // Version 4 fixes KAFKA-17011, which blocked SupportedFeatures.MinVersion in the response from being 0.
  //
  // Version 5 introduces ClusterId and NodeId checking and REBOOTSTRAP_REQUIRED error (KIP-1242).
  "validVersions": "0-5",
  "flexibleVersions": "3+",
  "headerVersions": {
    "0-2": "1",
    "3+": "2"
  },
  "fields": [
    { "name": "ClientSoftwareName", "type": "string", "versions": "3+",
      "ignorable": true, "about": "The name of the client." },
    { "name": "ClientSoftwareVersion", "type": "string", "versions": "3+",
      "ignorable": true, "about": "The version of the client." },
    { "name": "ClusterId", "type": "string", "versions": "5+", "nullableVersions": "5+", "default": "null",
      "ignorable": true, "about": "The cluster ID the client intends to connect to. Provide both ClusterId and NodeId, if known." },
    { "name": "NodeId", "type": "int32", "versions": "5+", "default": -1,
      "ignorable": true, "about": "The node ID the client intends to connect to. Provide both ClusterId and NodeId, if known." }
  ]

clients/src/main/resources/common/message/ApiVersionsRequest.json lines 17–43

Standard JSON has no comments, yet these files have // comments. That works because the generator reads them with Jackson's ALLOW_COMMENTS turned on (MessageGenerator.java line 187). The comments serve as a changelog of what each version changed. Reading the keys one by one:

The README.md nails down a few rules: "The order that fields appear in a message is important... Changing the order of the fields in a message is an incompatible change", "Protocol versions that have been released must be regarded as done. If there were mistakes, they should be corrected in a new version", and "Dropping support for old message versions is no longer allowed without a KIP" (a Kafka Improvement Proposal). Because each field carries its own versions, every version of a message lives together in one file.

A big request: Produce

The Produce request, which a producer uses to send records, looks like this.

  "fields": [
    { "name": "TransactionalId", "type": "string", "versions": "3+", "nullableVersions": "3+", "default": "null", "entityType": "transactionalId",
      "about": "The transactional ID, or null if the producer is not transactional." },
    { "name": "Acks", "type": "int16", "versions": "0+",
      "about": "The number of acknowledgments the producer requires the leader to have received before considering a request complete. Allowed values: 0 for no acknowledgments, 1 for only the leader and -1 for the full ISR." },
    { "name": "TimeoutMs", "type": "int32", "versions": "0+",
      "about": "The timeout to await a response in milliseconds." },
    { "name": "TopicData", "type": "[]TopicProduceData", "versions": "0+",
      "about": "Each topic to produce to.", "fields": [
      { "name": "Name", "type": "string", "versions": "0-12", "entityType": "topicName", "mapKey": true, "ignorable": true,
        "about": "The topic name." },
      { "name": "TopicId", "type": "uuid", "versions": "13+", "mapKey": true, "ignorable": true, "about": "The unique topic ID" },
      { "name": "PartitionData", "type": "[]PartitionProduceData", "versions": "0+",
        "about": "Each partition to produce to.", "fields": [
        { "name": "Index", "type": "int32", "versions": "0+",
          "about": "The partition index." },
        { "name": "Records", "type": "records", "versions": "0+", "nullableVersions": "0+",
          "about": "The record data to be produced." }
      ]}
    ]}
  ]

clients/src/main/resources/common/message/ProduceRequest.json lines 56–76

The settings you know from your application show up directly. The producer's acks setting becomes the Acks field (0, 1, -1), and using transactions fills in TransactionalId. One request holds several topics and each topic several partitions, a two-level array ([]PartitionProduceData inside []TopicProduceData), and the innermost Records, of type records, is the chunk of record batches the producer has gathered (part 5 builds that chunk, and in part 6 the broker receives and writes it).

The two-level array of a Produce request: one request → several topics → several partitions, with the records at the very inside. Fields and versions are taken from ProduceRequest.json above.

You can also see the version history. The topic name (Name) is used up to v12; from v13 it becomes a 16-byte topic ID (TopicId, KIP-516 in the comment). And validVersions is "3-13" (line 50), with flexible versions from v9 (line 51). As the comment at the top of the file says, v0–2 were removed in Kafka 4.0.

Easy to confuse: mapKey doesn't change the bytes

The "mapKey": true on Name and TopicId only tells the generated Java code to make that array "a set you can look up by this field" (the example's resp.apiKeys().find(key) is a method created that way). As the README says, the shape on the network stays the same, so it can be added or removed without bumping the version. Among the JSON keys, read the ones that change the bytes (versions, type, nullableVersions, tag) separately from the ones that only change the Java code (mapKey, about, entityType).

Example 2: a different API list per listener

To check that listeners is really used, we asked the same question (ApiVersions) on two ports of the same server. 9092 is the broker listener, 9093 the controller listener.

=== 4. the same question on each listener
port 9092: apiKeys=77 Produce=true Vote=false
port 9093: apiKeys=42 Produce=false Vote=true

The controller listener doesn't know Produce but does know Vote, KRaft's election request. Counting listeners in the JSON files gives 79 requests with "broker" and 42 with "controller". The controller side matches its 42 exactly; the broker side shows 77 because the two client-telemetry APIs (GetTelemetrySubscriptions, PushTelemetry) are dropped from the list when no telemetry exporter is configured (ApiVersionsResponse.java lines 196–199). What the controller does is covered in part 10.

3. How a schema becomes Java

Turning JSON into Java is the Gradle build's job — the processMessages task of the clients project.

  tasks.register('processMessages', JavaExec) {
    mainClass = "org.apache.kafka.message.MessageGenerator"
    classpath = configurations.generator
    args = [ "-p", "org.apache.kafka.common.message",
             "-o", "${projectDir}/build/generated/main/java/org/apache/kafka/common/message",
             "-i", "src/main/resources/common/message",
             "-t", "ApiMessageTypeGenerator",
             "-m", "MessageDataGenerator", "JsonConverterGenerator"
           ]

build.gradle lines 2155–2163

It runs the generator module's MessageGenerator as a Java program (JavaExec), reading the JSON in the -i folder and writing Java files into the -o folder. Further down the same block, compileJava.dependsOn 'processMessages' makes it run before compilation, and the generated folder is added to the source folders (lines 2186–2199). What Gradle tasks and dependencies are is covered in the Gradle series, and part 1 gave a first taste of this code-generation step in Kafka's build. Other modules such as metadata, raft and group-coordinator have a task with the same name that turns their own JSON (metadata records and so on) into Java.

The yellow box is the input people write; the green box is the output regenerated on every build.

MessageGenerator: read, check, generate

        List<ParsedSpec> parsedSpecs = new ArrayList<>();
        MessageSpec requestHeader = null;
        MessageSpec responseHeader = null;
        try (DirectoryStream<Path> directoryStream = Files.newDirectoryStream(Paths.get(inputDir), JSON_GLOB)) {
            for (Path inputPath : directoryStream) {
                try {
                    MessageSpec spec = JSON_SERDE.readValue(inputPath.toFile(), MessageSpec.class);
                    // ... (sets the two header schemas aside)
                    parsedSpecs.add(new ParsedSpec(inputPath, spec));
                // ...
        }
        // ...
        checkHeaderVersions(parsedSpecs, requestHeader, responseHeader);

        // Generate pass.
        for (ParsedSpec parsedSpec : parsedSpecs) {
            MessageSpec spec = parsedSpec.spec;
            try {
                outputFileNames.addAll(
                    generateAndWriteMessageClasses(spec, packageName, outputDir, messageClassGeneratorTypes));
                numProcessed++;
                typeClassGenerators.forEach(generator -> generator.registerMessageType(spec));

generator/src/main/java/org/apache/kafka/message/MessageGenerator.java lines 254–293

1. Read: Jackson turns each JSON into a MessageSpec object. This step also sets the two header schemas aside.
2. Check: it verifies that every file's headerVersions points at versions that really exist in the header schemas. The comments in the elided spots (lines 250–253 and 281–283) give the reason: a single file can't reach the header schemas, so this has to wait until everything is read, and a bad schema fails the build before anything is generated, so no half-generated output is left behind.
3. Generate: for each file it runs the generators given with -m (MessageDataGenerator, JsonConverterGenerator) and registers every message with the -t generator (ApiMessageTypeGenerator). At the end ApiMessageTypeGenerator writes one ApiMessageType.java holding the whole API list (lines 298–301). Old files not produced by this run are deleted (lines 303–312).

Each JSON field object is unpacked by Jackson into the arguments of the FieldSpec constructor (@JsonProperty("versions") String versions, @JsonProperty("tag") Integer tag …, FieldSpec.java lines 62–75). Version strings like "3+", "0-12" and "none" are turned into a [lowest, highest] pair by Versions.parse(); "3+" sets the upper end to Short.MAX_VALUE (Versions.java lines 40–65).

A version range becomes if (_version >= 3)

The core idea of the generator is simple: print a field's version range as an if statement. It shows clearly in generateVariableLengthReader(), which prints the code that reads fields with a length, such as strings and arrays.

        String lengthVar = type.isArray() ? "arrayLength" : "length";
        buffer.printf("int %s;%n", lengthVar);
        VersionConditional.forVersions(fieldFlexibleVersions, possibleVersions).
            ifMember(__ -> buffer.printf("%s = _readable.readUnsignedVarint() - 1;%n", lengthVar)).
            ifNotMember(__ -> {
                if (type.isString()) {
                    buffer.printf("%s = _readable.readShort();%n", lengthVar);
                } else if (type.isBytes() || type.isArray() || type.isRecords()) {
                    buffer.printf("%s = _readable.readInt();%n", lengthVar);
                } else {
                    throw new RuntimeException("Can't handle variable length type " + type);
                }
            }).
            generate(buffer);

generator/src/main/java/org/apache/kafka/message/MessageDataGenerator.java lines 630–643

VersionConditional is a helper that prints "this code if the version is in this range (fieldFlexibleVersions), that code otherwise" as if (_version >= N). If the range covers all the versions being handled, it prints only one side with no if. So flexible versions get code that reads the length as a varint (readUnsignedVarint() - 1), and other versions read 2 bytes (readShort()) for strings or 4 bytes (readInt()) for arrays. Here is the start of the ApiVersionsRequestData.read() printed that way.

    @Override
    public final void read(Readable _readable, short _version) {
        if (_version >= 3) {
            int length;
            length = _readable.readUnsignedVarint() - 1;
            if (length < 0) {
                throw new RuntimeException("non-nullable field clientSoftwareName was serialized as null");
            } else if (length > 0x7fff) {
                throw new RuntimeException("string field clientSoftwareName had invalid length " + length);
            } else {
                this.clientSoftwareName = _readable.readString(length);
            }
        } else {
            this.clientSoftwareName = "";
        }

[Lines 115–129 of clients/build/generated/main/java/org/apache/kafka/common/message/ApiVersionsRequestData.java, generated from ApiVersionsRequest.json at build time (generated at the reference commit) — original schema: ApiVersionsRequest.json lines 35–36]

The JSON's "versions": "3+" becomes the outer if (_version >= 3), and versions without the field get the default (""). Since ClientSoftwareName's version range lies within flexibleVersions (3+), only the varint read is left inside. The field can't be null, so a negative length throws. The generator prints this kind of if chain — easy to get wrong by hand, version after version — for every message.

The topic entry of a Produce request uses entirely different fields depending on the version.

        public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
            int _numTaggedFields = 0;
            if (_version <= 12) {
                {
                    byte[] _stringBytes = _cache.getSerializedValue(name);
                    if (_version >= 9) {
                        _writable.writeUnsignedVarint(_stringBytes.length + 1);
                    } else {
                        _writable.writeShort((short) _stringBytes.length);
                    }
                    _writable.writeByteArray(_stringBytes);
                }
            }
            if (_version >= 13) {
                _writable.writeUuid(topicId);
            }

[TopicProduceData.write() in ProduceRequestData.java, generated from ProduceRequest.json at build time, lines 607–622 (generated at the reference commit) — original schema: ProduceRequest.json lines 65–67]

"versions": "0-12" became if (_version <= 12) and "13+" became if (_version >= 13). Inside, the way the length is written splits again on whether the version is flexible (9+). In the same file, PartitionProduceData.write() hands the records over with _writable.writeRecords(records) (line 949), and addSize() counts their size separately with addZeroCopyBytes() (line 977). This is where section 1's "records are not copied into the buffer" begins.

The generated …Data classes only move between bytes and fields; hand-written wrappers like ApiVersionsRequest handle checks (isValid()) and building error responses.

• Every generated class implements the Message interface (size()/addSize(), write(), read(), unknownTaggedFields(), duplicate()), and requests, responses and headers implement ApiMessage, which adds apiKey(). A header's apiKey() returns -1.
• Bytes are read and written through the Readable/Writable interfaces; the common implementation is ByteBufferAccessor, which wraps a ByteBuffer. Section 1's SendBuilder is also a Writable.
• Classes like ApiVersionsRequest and ProduceRequest in the common.requests package are hand-written wrappers that return the generated object from data() (the AbstractRequestResponse interface).

At this commit the clients generated folder holds 391 files and 186,401 lines (counted with ls | wc -l and cat *.java | wc -l): 195 generated message classes, 195 …JsonConverters that turn the same messages into JSON text, and one ApiMessageType. There are 203 JSON files but only 195 message classes because no classes are generated for the 8 requests/responses of the four old APIs with no versions left ("validVersions": "none"); they keep only a placeholder with empty schemas in ApiMessageType.

Easy to confuse: what happens to a value when you write a version without that field?

When you write a message at an older version that lacks a field, the value you put in that field doesn't make it into the bytes. Here ignorable decides. MessageDataGenerator prints a check that throws UnsupportedVersionException "if the value isn't the default" only for fields that are not ignorable (lines 815–818, FieldSpec.java lines 649–662). The generated AddPartitionsToTxnRequestData contains lines like "Attempted to write a non-default transactions at version " for this reason. All four ApiVersions fields, on the other hand, are ignorable, so the generated write() has no code checking ClusterId below v5. Losing a client name on its way to an old broker is fine, but a value that changes the meaning of a request must not be dropped silently — the README's example is a "verify only" boolean.

Example 3: the same request, bytes of two versions

We serialized one ApiVersions request as v2 (not flexible) and v3 (flexible) and printed the bytes (bytes mode). The body carries the client name u03-app and version 1.0.

=== 2. request bytes: ApiVersions v2 vs v3
ApiVersions v2 request, header v1, 17 bytes:
  00 00 00 0d 00 12 00 02 00 00 00 01 00 03 75 30 33
ApiVersions v3 request, header v2, 31 bytes:
  00 00 00 1b 00 12 00 03 00 00 00 02 00 03 75 30 33 00 08 75 30 33 2d 61 70 70 04 31 2e 30 00

v2 is 17 bytes: after the length (13) there's only a header and no body, because, per the JSON, v0–2 have no fields at all. We did set a client name, but it's ignorable, so it was silently dropped. v3 is 31 bytes; split byte by byte, it looks like the figure below.

Blue boxes are the request header, green boxes the body. Note that only the header's ClientId uses the old encoding (a 2-byte length) — see section 4.

• The header's ClientId is 00 03 + u03, a 2-byte length, while the body's strings are 08 + u03-app, a varint (length+1). One request uses two ways of writing a string; section 4 explains why.
• Flexible versions add one tagged-field count byte (00) at the end of the header and one at the end of the body (section 6).

4. The request header and its version

The request header is defined in JSON too.

  "validVersions": "1-3",
  "flexibleVersions": "2+",
  "fields": [
    { "name": "RequestApiKey", "type": "int16", "versions": "0+",
      "about": "The API key of this request." },
    { "name": "RequestApiVersion", "type": "int16", "versions": "0+",
      "about": "The API version of this request." },
    { "name": "CorrelationId", "type": "int32", "versions": "0+",
      "about": "The correlation ID of this request." },

    // The ClientId string must be serialized with the old-style two-byte length prefix.
    // The reason is that older brokers must be able to read the request header for any
    // ApiVersionsRequest, even if it is from a newer version.
    // Since the client is sending the ApiVersionsRequest in order to discover what
    // versions are supported, the client does not know the best version to use.
    { "name": "ClientId", "type": "string", "versions": "1+", "nullableVersions": "1+", "flexibleVersions": "none",
      "about": "The client ID, for identifying an application." },
    { "name": "ClientInstanceId", "type": "uuid", "versions": "3+", "taggedVersions": "3+", "tag": 0,
      "about": "The client instance ID, for identifying an instance of an application." }
  ]

clients/src/main/resources/common/message/RequestHeader.json lines 28–47

The first three fields (API key, version, correlation ID) have the same position and size in every version. That is the promise the whole protocol rests on: the receiver can tell which request and which version it has by reading just the first 4 bytes.

ClientId carries a field-level "flexibleVersions": "none". Even though header v2 is a flexible version, this one field is written the old way (a 2-byte length) — the 00 03 75 30 33 we saw in the bytes of Example 3. The comment gives the reason: the client sends ApiVersions without knowing which versions the broker supports, so even an old broker must be able to read that request's header.

The header version comes from the request version — chicken and egg

There is no field for the header version. The header version is determined by the request's API key and version, and those two live inside the header. The broker solves this as follows.

    public static RequestHeader parse(ByteBuffer buffer) {
        short apiKeyId = -1;
        try {
            // We derive the header version from the request api version, so we read that first.
            // The request api version is part of `RequestHeaderData`, so we reset the buffer position after the read.
            int bufferStartPositionForHeader = buffer.position();
            apiKeyId = buffer.getShort();
            short apiVersion = buffer.getShort();
            ApiKeys apiKey = ApiKeys.forId(apiKeyId);

            // `apiKey.requestHeaderVersion` will fail if there are no valid versions - we do this check first in order to
            // provide a more helpful message
            if (!apiKey.hasValidVersion())
                throw new InvalidRequestException("Unsupported api with key " + apiKeyId + " (" + apiKey.name + ") and version " + apiVersion);

            short headerVersion = apiKey.requestHeaderVersion(apiVersion);
            buffer.position(bufferStartPositionForHeader);
            final RequestHeaderData headerData = new RequestHeaderData(new ByteBufferAccessor(buffer), headerVersion);

clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java lines 140–157

It peeks at the first 2+2 bytes (two getShort() calls), decides the header version, rewinds (buffer.position(...)) and reads everything again from the start with the generated RequestHeaderData. This only works because the first two fields have the same shape in every version.

To know the header version you need two fields inside the header. Because those two fields are always first and always the same size, "peek, rewind, reread" works.

The mapping table: from a rule in 4.4.0 to JSON in trunk

requestHeaderVersion() is a method of the generated ApiMessageType. Inside, under switch (apiKey), each API has an if chain such as case 0: // Produce → if (_version >= 9) { return (short) 2; } else { return (short) 1; } (from line 769 of ApiMessageType.java, generated at build time at the reference commit — the source is each request JSON's headerVersions, e.g. ProduceRequest.json lines 52–55).

What this table is built from is where this trunk commit differs from the widely used 4.4.0.

• 4.4.0: the JSON has no headerVersions key. ApiMessageTypeGenerator hard-codes the rule "if the request version is flexible, header v2, otherwise v1 / for responses v1 or v0", and treats only the ApiVersions response specially with apiKey == 18 (checked in 4.4.0's ApiMessageTypeGenerator.generateHeaderVersion()).
• trunk (this commit): each request and response JSON spells out its headerVersions table, and the generator's generateHeaderVersionFromMap() just intersects that table with the valid version range and copies it into ifs (ApiMessageTypeGenerator.java lines 384–411). This is also why steps 1 and 2 of the generator in section 3 check this table against the header schemas.

What looks like the reason for the change is in the same commit: the request header gained a v3 (4.4.0 has "validVersions": "1-2"). v3 adds a tagged field called ClientInstanceId (KIP-1313 in the comment), and right now only v1 of the GetTelemetrySubscriptions, PushTelemetry and OffsetDelete requests uses header v3 (all three mark their last version as "not yet stable" with "latestVersionUnstable": true, so a broker with default settings doesn't advertise that v1 — see "What this post leaves out" at the end). A single rule like "flexible means v2" can't express that mapping. We did not confirm the motivation from the commit history or the KIP; the above comes from comparing the source of the two versions.

Easy to confuse: the ApiVersions response header is always v0

ApiVersionsResponse.json has only one mapping, "0+": "0" (lines 38–40). Even though v3 and up are flexible, the response header is v0, with no tagged-field slot. The file's comments cite KIP-511 and say "The length of the header must not change", because even a client that doesn't know the version must be able to read the response's correlation ID. The README also names this response as the exception to the rule that a flexible version must map to header version 1 or later (for responses).

Example 4: the first 12 bytes of a v2 and a v4 response

The versions mode asks ApiVersions on one connection, once as v2 and once as v4, and prints the start of each response.

=== 3. ApiVersions over a raw socket: v2 vs v4
--- v2: response 476 bytes (after the 4-byte length), first 12: 00 00 00 01 00 00 00 00 00 4d 00 00
correlationId=1 errorCode=0 apiKeys=77 supportedFeatures=0 finalizedFeaturesEpoch=-1
  Produce(0): 0 to 13
  Fetch(1): 4 to 18
  Metadata(3): 0 to 13
  ApiVersions(18): 0 to 5
--- v4: response 895 bytes (after the 4-byte length), first 12: 00 00 00 02 00 00 4e 00 00 00 00 00
correlationId=2 errorCode=0 apiKeys=77 supportedFeatures=7 finalizedFeaturesEpoch=49
  Produce(0): 0 to 13
  Fetch(1): 4 to 18
  Metadata(3): 0 to 13
  ApiVersions(18): 0 to 5
  feature group.version: 0 to 1
  feature kraft.version: 0 to 1
  feature metadata.version: 7 to 33
  feature share.version: 0 to 2
  feature streams.version: 0 to 1
  feature transaction.version: 0 to 2
  feature eligible.leader.replicas.version: 0 to 1

The finalizedFeaturesEpoch value differs from run to run. Splitting the first 12 bytes:

1. In both responses ErrorCode comes right after the correlation ID. The v4 response, too, has the header v0 described above, with no tag-count byte in the header.
2. The array length is 4 bytes in v2 (0x4d = 77) and a 1-byte varint in v4 (0x4e = 78 = 77+1) — exactly the two branches printed by section 3's generateVariableLengthReader().
3. Only the v4 response carries 7 SupportedFeatures. In the JSON this field is a tagged field with "tag": 0, "taggedVersions": "3+", so v2 has no room for it (section 6). The response grows from 476 to 895 bytes mostly because of the feature lists carried as tagged fields (SupportedFeatures, FinalizedFeatures); in flexible versions each of the 77 array entries also gets its own tag-count byte.

5. Version negotiation — ApiVersions

Clients and brokers can talk across different Kafka versions because, right after connecting, they first ask "which versions of which APIs do you know?". That question is the ApiVersions request (API key 18).

The numbers match the order of the explanation below. The queue and threads the request passes in step 4 are covered in part 4.

Client: ask first, as soon as the connection is up

            if (discoverBrokerVersions) {
                nodesNeedingApiVersionsFetch.put(node, new ApiVersionsRequest.Builder());
                log.debug("Completed connection to node {}. Fetching API versions.", node);
            } else {
                this.connectionStates.ready(node);
                log.debug("Completed connection to node {}. Ready.", node);
            }

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java lines 1201–1207 (handleConnections())

An established TCP connection does not become "ready" right away (①). Once the node is put on the list of nodes needing ApiVersions, the very next step of the same poll() (handleInitiateApiVersionRequests(), line 733) switches the connection state to CHECKING_API_VERSIONS and sends the request when the channel is ready (lines 1211–1240). The builder new ApiVersionsRequest.Builder() allows oldestVersion() to latestVersion(), that is, 0 to 5 (ApiVersionsRequest.java lines 40–44).

Which version does this request go out with? The version is decided in doSend().

            NodeApiVersions versionInfo = apiVersions.get(nodeId);
            short version;
            // Note: if versionInfo is null, we have no server version information. This would be
            // the case when sending the initial ApiVersionRequest which fetches the version
            // information itself.  It is also the case when discoverBrokerVersions is set to false.
            if (versionInfo == null) {
                version = builder.latestAllowedVersion();
                // ...
            } else {
                version = versionInfo.latestUsableVersion(clientRequest.apiKey(), builder.oldestAllowedVersion(),
                        builder.latestAllowedVersion());
            }
            // The call to build may also throw UnsupportedVersionException, if there are essential
            // fields that cannot be represented in the chosen version.
            doSend(clientRequest, isInternalRequest, now, builder.build(version));

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java lines 641–657

With no version info for that node yet (versionInfo == null), it sends the highest version it knows (②, v5 for ApiVersions). Once version info exists, NodeApiVersions.latestUsableVersion() chooses (⑧⑨, NodeApiVersions.java lines 149–165). It uses the highest version (maxVersion()) in the overlap between the range the broker knows, [min, max], and the range the client can send (ApiVersionsResponse.intersect() takes the larger min and the smaller max, lines 306–320). With no overlap, it doesn't even send the request and ends with UnsupportedVersionException. The Produce builder the producer uses covers ApiKeys.PRODUCE.oldestVersion() to latestVersion(), that is, 3 to 13 (lowering the top to 11 only when using the transaction v1 protocol, ProduceRequest.java lines 45–51).

When the response arrives, handleApiVersionsResponse() builds a NodeApiVersions, puts it into ApiVersions (a map from node ID to NodeApiVersions), and marks the connection "ready" (⑥⑦, lines 1161–1167). Version info is kept per node, because while a cluster's brokers are upgraded one by one (a rolling upgrade), different brokers can know different versions.

Broker: an ApiVersions in an unknown version gets a v0 answer

If the client is newer than the broker, the broker may not know the version of the first ApiVersions (v5). For any other request an unknown version is simply an error (Example 6 below), but ApiVersions gets special treatment.

First, ApiKeys.isVersionEnabled() lets any version through for ApiVersions with one line, if (this == ApiKeys.API_VERSIONS) return true;. The comment says "The client always send the highest version that it supports and the server fails back to version 0 if it does not know it" (ApiKeys.java lines 246–253).

Then, when decoding the body, if it is "an ApiVersions in an unknown version" (isUnsupportedApiVersionsRequest()), RequestContext.parseRequest() doesn't read the body at all and substitutes new ApiVersionsRequest(new ApiVersionsRequestData(), (short) 0, header.apiVersion()), an empty v0 request (RequestContext.java lines 112–117). The body of an unknown version can't be decoded, since nobody knows its shape. The response is serialized as v0 too (apiVersion() returns 0 in this case, lines 158–163). KafkaApis does the handling.

    def createResponseCallback(requestThrottleMs: Int): ApiVersionsResponse = {
      val apiVersionRequest = request.body(classOf[ApiVersionsRequest])
      if (apiVersionRequest.hasUnsupportedRequestVersion) {
        apiVersionRequest.getErrorResponse(requestThrottleMs, Errors.UNSUPPORTED_VERSION.exception)
      } else if (!apiVersionRequest.isValid) {
        apiVersionRequest.getErrorResponse(requestThrottleMs, Errors.INVALID_REQUEST.exception)
      } else if (clusterIdOrNodeIdIsInvalid(apiVersionRequest)) {
        apiVersionRequest.getErrorResponse(requestThrottleMs, Errors.REBOOTSTRAP_REQUIRED.exception)
      } else {
        apiVersionManager.apiVersionResponse(requestThrottleMs, request.header.apiVersion() < 4)
      }
    }

core/src/main/scala/kafka/server/KafkaApis.scala lines 1551–1562

Reading the Scala: def f(x: Int): T = { … } is a method returning T, and, as here, a method can be defined inside another method. val is a variable that can't be reassigned (Java's final). classOf[ApiVersionsRequest] is Java's ApiVersionsRequest.class. In Scala, if … else is an expression that produces a value, so without any return, the value of the chosen branch becomes the method's return value.

For an unknown version it builds an UNSUPPORTED_VERSION error response. Here getErrorResponse() doesn't stop at the error code; it also includes the range of ApiVersions versions the broker knows (ApiVersionsRequest.java lines 134–140, KIP-511). On receiving that response the client does not disconnect; it asks again with the maximum it was given.

            } else {
                // Starting from Apache Kafka 2.4, ApiKeys field is populated with the supported versions of
                // the ApiVersionsRequest when an UNSUPPORTED_VERSION error is returned.
                // If not provided, the client falls back to version 0.
                short maxApiVersion = 0;
                if (!apiVersionsResponse.data().apiKeys().isEmpty()) {
                    ApiVersion apiVersion = apiVersionsResponse.data().apiKeys().find(ApiKeys.API_VERSIONS.id);
                    if (apiVersion != null) {
                        maxApiVersion = apiVersion.maxVersion();
                    }
                }
                nodesNeedingApiVersionsFetch.put(node, new ApiVersionsRequest.Builder(maxApiVersion));
            }

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java lines 1146–1158

The branch just above (lines 1141–1145) disconnects when even a v0 request got an error, or when the error isn't UNSUPPORTED_VERSION.

This works because the response header is fixed at v0 and the shape of the v0 response body (ErrorCode, the ApiKeys array) has never changed. Section 4's "the ApiVersions response header is always v0" and "ClientId uses the old encoding" both exist for this one round trip.

Example 5: asking with a version newer than the broker

The too-new mode sends ApiVersions as v99 and decodes the response as v0.

=== 5. ApiVersions v99 (newer than the broker)
errorCode=35 (UNSUPPORTED_VERSION) apiKeys=1
  ApiVersions(18): 0 to 5

Along with error code 35 (UNSUPPORTED_VERSION, Errors.java line 265), the list holds only ApiVersions, "0 to 5". A real client would now ask again with v5.

Example 6: other requests in an unknown version

The same experiment with a Metadata request (API key 3) ends differently. The metadata-too-new mode sends Metadata v99.

=== 6. Metadata v99 (newer than the broker)
no response: the broker closed the connection
Closing socket for <connection id> because of error
UnsupportedVersionException: Received request for api with key 3 (Metadata) and unsupported version 99

The last two lines were grepped by run.sh from the broker's server.log; the connection ID differs per run, so it was replaced with <connection id>. No response came; the connection was closed, because the broker's network thread checks the version as soon as it decodes the header.

    public static RequestHeader parseRequestHeader(ApiVersionManager apiVersionManager, ByteBuffer buffer) {
        RequestHeader header = RequestHeader.parse(buffer);
        if (apiVersionManager.isApiEnabled(header.apiKey(), header.apiVersion())) {
            return header;
        } else if (header.isApiVersionSupported()) {
            throw new InvalidRequestException("Received request for disabled api with key " + header.apiKey().id + " (" + header.apiKey().name() + ") and version " + header.apiVersion());
        } else {
            throw new UnsupportedVersionException("Received request for api with key " + header.apiKey().id + " (" + header.apiKey().name + ") and unsupported version " + header.apiVersion());
        }
    }

server/src/main/java/org/apache/kafka/network/Processor.java lines 33–42

isApiEnabled() checks both "does this listener accept this API" (inScope) and "is this a known version" (isVersionEnabled) (ApiVersionManager.java lines 62–64). The exception is caught in processCompletedReceives() in SocketServer.scala, and processChannelException() closes the connection. The body of an unknown version has an unknown shape, so the broker can't even build an error response in that version. This is why a client must not skip negotiation and send whatever version it likes.

The names are easy to mix up: server/.../network/Processor.java is a small Java class with a few helpers like this one, while the network thread itself, Processor, is a Scala class inside SocketServer.scala. Its processCompletedReceives() calls this parseRequestHeader() for each received payload (line 989; the file imports the Java class under another name, JProcessor), builds a RequestContext and a Request, and sends it to the request queue (SocketServer.scala lines 982–1020). And inside the Request constructor, context.parseRequest(buffer) decodes the body as well (Request.java line 101). In short, the broker's network thread decodes the header and body with the generated classes, and the request handler threads receive objects that are already decoded. The sendRequest() and mute() that follow are part 4's story.

The path to the red box gets no response. ApiVersions is the only request that can be answered in an unknown version.

The switch in AbstractRequest.parseRequest() lists the 91 APIs that still have versions one by one, by hand (case PRODUCE: return ProduceRequest.parse(readable, apiVersion); …, from AbstractRequest.java line 177). Each parse() just calls the generated …Data(readable, version) constructor and puts the result in a wrapper (e.g. ProduceRequest.java lines 234–236).

Example 7: looking with the tool, and a real producer's conversation

The Kafka distribution ships a tool that asks ApiVersions, kafka-broker-api-versions.sh.

=== 1. kafka-broker-api-versions.sh (selected lines)
localhost:9092 (id: 1 rack: null isFenced: false) -> (
	Produce(0): 0 to 13 [usable: 13],
	Fetch(1): 4 to 18 [usable: 18],
	Metadata(3): 0 to 13 [usable: 13],
	ApiVersions(18): 0 to 5 [usable: 5],
lines with an api: 80
	GetTelemetrySubscriptions(71): UNSUPPORTED,
	PushTelemetry(72): UNSUPPORTED,

[usable: N] is the version this tool (a client from the same commit) would use, i.e. min(client max, broker max) (NodeApiVersions.java lines 236–237). The 80 lines containing a parenthesis are, minus the first line, 79 APIs: the 77 the broker reported plus 2 the client knows but the broker's response lacks (UNSUPPORTED; toString() fills these in by walking ApiKeys.clientApis(), lines 194–200). Why the two telemetry APIs are missing we saw in Example 2.

Easy to confuse: why is Produce "0 to 13"?

The JSON says Produce has "validVersions": "3-13", yet the broker answered "0 to 13". It does so on purpose.

        // see `PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION` for details on why we do this
        short oldestVersion = (this == PRODUCE && listenerType.map(l -> l == ApiMessageType.ListenerType.BROKER).orElse(false)) ?
            PRODUCE_API_VERSIONS_RESPONSE_MIN_VERSION : oldestVersion();

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java lines 289–291

The comment above the constant (lines 157–160) explains: v0–2 were removed in Kafka 4.0, but because of a bug in librdkafka, the C client library (KAFKA-18659), the ApiVersions response still advertises 0 as the minimum. Actually sending a v0–2 Produce gets it rejected, as the comment at the top of the JSON says. It's an example of ApiVersions numbers not always matching the versions the broker accepts.

How does a real producer hold this conversation? Negotiate.java sends one record with a plain KafkaProducer, and a client logging config (log4j2-client.yaml) keeps only NetworkClient at DEBUG.

=== 9. a real producer: NetworkClient DEBUG log
sent to u03-0 at offset 0
--- the first request header, as the client logs it
RequestHeader(apiKey=API_VERSIONS, apiVersion=5, clientId=u03-producer, clientInstanceId=AAAAAAAAAAAAAAAAAAAAAA, correlationId=0, headerVersion=2)
--- connections and requests in order (shortened)
Completed connection to node -1. Fetching API versions.
  -> API_VERSIONS v5 (header v2, correlationId=0) to node -1
  <- API_VERSIONS v5 (correlationId=0) from node -1
  -> METADATA v13 (header v2, correlationId=1) to node -1
  -> INIT_PRODUCER_ID v5 (header v2, correlationId=2) to node -1
  <- METADATA v13 (correlationId=1) from node -1
  <- INIT_PRODUCER_ID v5 (correlationId=2) from node -1
Completed connection to node 1. Fetching API versions.
  -> API_VERSIONS v5 (header v2, correlationId=4) to node 1
  <- API_VERSIONS v5 (correlationId=4) from node 1
  -> INIT_PRODUCER_ID v5 (header v2, correlationId=5) to node 1
  <- INIT_PRODUCER_ID v5 (correlationId=5) from node 1
  -> PRODUCE v13 (header v2, correlationId=6) to node 1
  <- PRODUCE v13 (correlationId=6) from node 1

The log lines are long, so run.sh shortened them with sed (-> is a Sending … request line, <- a Received … response line).

1. ApiVersions comes first on every connection. At first the client connects to the bootstrap.servers address without knowing the node ID, so it uses a temporary ID, -1. After the Metadata response tells it that broker is node 1, it opens a new connection as node 1 — same address — and again starts with ApiVersions.
2. Versions follow the negotiation. ApiVersions is v5, Metadata and Produce are v13. On this server, client and broker maximums are equal for both.
3. Header v2, and clientInstanceId. The logged RequestHeader shows clientInstanceId, new in this commit, but its value is zero (AAAA… is the base64 of the zero UUID) and the header version is 2. This field is a tagged field of header v3, so it isn't carried in v2 (the RequestHeader constructor only sets it when headerVersion >= 3, RequestHeader.java lines 51–54).
4. Before the Metadata response arrived, InitProducerId was sent on the same connection. As section 1 showed, the responses came back in the order sent. Correlation ID 3 doesn't appear in this log; we did not chase what took that number.

InitProducerId is how an idempotent producer (the default for enable.idempotence is true, ProducerConfig.java lines 574–576) gets a producer ID. Running the same example once more and reading the raw log showed that the first InitProducerId response carried error code 14 (COORDINATOR_LOAD_IN_PROGRESS), which is why it was sent again. What this request means is covered in part 12.

6. Flexible versions and tagged fields

"Flexible versions" have come up repeatedly in sections 3 and 4. They are the versions within a JSON file's flexibleVersions range, and a generator comment calls them "a KIP-482 flexible schema" (SchemaGenerator.java line 72). Two things change in flexible versions.

The first is compactness. The length of a short string or array almost always fits in 1 byte — the ApiKeys array length shrinking from 4 bytes to 1 in Example 4. The "+1" lets 0 stand for null (the old encoding used -1 for null).

The second, and more important, is tagged fields. In flexible versions every struct ends with an "extras slot". It can hold any number of (tag number, size, bytes) groups, and the receiver decodes the tags it knows and skips the ones it doesn't, by their size. So tagged fields can be added to an existing version without bumping it. In the README's words, "Older servers will ignore new tagged fields which they do not understand."

The end of the generated ApiVersionsRequestData.read() is that "skipping".

        this._unknownTaggedFields = null;
        if (_version >= 3) {
            int _numTaggedFields = _readable.readUnsignedVarint();
            // ...
            for (int _i = 0; _i < _numTaggedFields; _i++) {
                int _tag = _readable.readUnsignedVarint();
                int _size = _readable.readUnsignedVarint();
                switch (_tag) {
                    default:
                        this._unknownTaggedFields = _readable.readUnknownTaggedField(this._unknownTaggedFields, _tag, _size);
                        break;
                }
            }
        }

[Lines 161–182 of ApiVersionsRequestData.java, generated from ApiVersionsRequest.json at build time (generated at the reference commit) — original schema: ApiVersionsRequest.json line 29]

The ApiVersions request defines no tagged fields, so the switch has only default. Unknown tags aren't thrown away; their raw bytes are collected in _unknownTaggedFields (Readable.readUnknownTaggedField()). If the message is written again, those bytes go out again (RawTaggedFieldWriter, lines 214–218 of the generated write()). A message that defines tagged fields gets branches like case 0:. That is how the ApiVersionsResponse fields seen in Example 4 — SupportedFeatures (tag 0), FinalizedFeaturesEpoch (tag 1) and FinalizedFeatures (tag 2) — are read.

Example 8: sending a tag the broker doesn't know

The unknown-tag mode attaches tag 99 (content hello), which this broker doesn't know, to the body of an ApiVersions v4 request. Then it tries to serialize the same message as the non-flexible v2.

// An example written for this post (not part of Kafka's source)
                    ApiVersionsRequestData data = body("u03-app", "1.0");
                    data.unknownTaggedFields().add(new RawTaggedField(99, "hello".getBytes(StandardCharsets.UTF_8)));
                    out.write(frame((short) 4, data));

[Abridged from the example ex/u03-wire-protocol/RawWire.java]

=== 7. an unknown tagged field
v4 with tag 99: errorCode=0 apiKeys=77
v2 with tag 99: UnsupportedVersionException: Tagged fields were set, but version 2 of this message does not support them.

1. The v4 request got the usual list of 77 with no error (errorCode=0). The broker skipped tag 99. A new client can put new information in the extras slot without breaking an old broker.
2. Writing it as v2 failed on the client, before anything was sent, because non-flexible versions have no extras slot. This is the if (_version >= 3) { … } else { if (_numTaggedFields > 0) throw … } branch of the generated write() (lines 219–223).

An unknown tag is skipped by its size. The bytes of the first two fields are the same as in section 3's figure (v3); the tagged-area bytes are computed from the order RawTaggedFieldWriter writes them (tag, size, bytes). The two result lines are example 8's actual output.

Easy to confuse: tagged fields can't appear in just any version

Tagged fields can be added without bumping the version, but only to flexible versions (README: 'tagged fields can only be added to "flexible" message versions'). The generator stops the build if a field's taggedVersions doesn't fall within the flexible range (MessageDataGenerator.java lines 436–440). And a tag number, once used, can never be reused for another meaning, because old clients would decode it with the old meaning. Old versions that predate flexible versions have no extras slot, so adding anything to them still requires a new version. Flexible versions start at a different point for each API — v9 for Produce, v12 for Fetch.

What this post leaves out

• …JsonConverter, which turns messages into JSON text (used by server/src/main/java/org/apache/kafka/network/RequestConvertToJson.java): generator/src/main/java/org/apache/kafka/message/JsonConverterGenerator.java
• The older schema objects Schema, Field, Type (the SCHEMA_0 … constants in generated classes): clients/src/main/java/org/apache/kafka/common/protocol/types/
• latestVersionUnstable, which marks the last version "not yet stable", and the broker setting unstable.api.versions.enable (ServerConfigs.java line 124). At this commit four files use it, including InitProducerIdRequest. It looks related to Example 7's InitProducerId going out as v5 rather than v6, but we didn't follow it
• Envelope requests that forward requests to another broker, and forwardable: common/requests/EnvelopeRequest.java, forwardable in ApiKeys.java
• Cluster feature versions (metadata.version and the rest of Example 4's feature list) and their negotiation: server-common/src/main/java/org/apache/kafka/server/common/Feature.java (continued in part 10)
• KIP-1242's ClusterId/NodeId check and REBOOTSTRAP_REQUIRED: clusterIdOrNodeIdIsInvalid() at line 1567 of KafkaApis.scala, lines 1130–1140 of NetworkClient.java
• Requests during SASL authentication (SaslHandshake, SaslAuthenticate) and the reserved correlation ID range: common/security/authenticator/SaslClientAuthenticator.java
• JSON schemas that aren't requests, like metadata records and group coordinator records (same generator, other modules): metadata/src/main/resources/common/metadata/, group-coordinator/src/main/resources/common/message/
• Left out of the whole series: Kafka Streams, Kafka Connect, MirrorMaker, share groups, tiered storage, security (SASL, TLS, ACLs), quotas, the Admin client, JBOD

Summary

What's next

We've followed the bytes until they become a Request object on the broker's network thread. Next we look at that network thread itself: the Acceptor that accepts connections, the Processor that watches many connections with a selector and calls this post's parseRequestHeader(), the request handler threads beyond the request queue that call KafkaApis.handle(), and mute/unmute, which keeps "responses in the order sent". It continues in part 4: The Broker's Network Layer — Acceptors, Processors and Request Handlers.

An HTTP server solves the same "bytes into a request object" problem by scanning text. Put Tomcat part 6's Http11InputBuffer, reading the request line and headers character by character, next to this post's NetworkReceive, which reads the length first, and you can see what a binary protocol gains and gives up. HTTP, too, switched from text to binary chunks called frames in HTTP/2 (Tomcat part 15). For another example of producing Java code from definition files, see Tomcat part 13: Jasper, which turns JSPs into servlet Java files.

• Part 2: The Big Picture — Clients, Brokers, Controllers and Partitions
• Part 5: Inside the Producer — From send() to Batches and the Sender Thread
• Part 6: How the Broker Accepts Records — The Record Batch Format and the Produce Path
• Digging into Gradle's Source 1: How I Read It with AI, and the Map