feat(protocol): negotiate API versions

[trapped]

This PR introduces basic Kafka API version negotiation in Sarama by enabling the client to restrict the API version for each request based on the version ranges advertised by brokers in the initial ApiVersions request/response - which is now executed synchronously as part of the broker connection process, rather than as a fire-and-forget deferred goroutine, and before SASL authentication takes place (but after SSL/TLS, as indicated by the Kafka protocol docs).

The selected version is the highest supported by both the broker and the user-defined Config.Version upper bound.

To support this dynamic selection, a new setVersion(int16) method has been added to the protocolBody interface, exposing the request/response structs' Version field to be set externally.

Also, it defines constants for message keys, and replaces some (read: those I found) usages of numeric literals to use the constants instead.

I experimented with extracting more logic from protocolBody implementations, too, such as centralizing requiredVersion(), isValidVersion(), and the "max protocol version for Kafka version" selection in most Request constructors as suggested by @dnwe in my previous PR; however, while they can certainly simplify the code, I found these were significantly larger-scale changes (and kinda out of scope for the purpose of this PR), and decided to leave them out to keep this PR short and easy to review.

Motivation behind the change: Sarama incompatibility with Redpanda

The latest github.com/IBM/sarama version fails to connect to Redpanda brokers (latest version too):

• Redpanda rejects the request with Unsupported version 10 for metadata API
• sarama fails with kafka: client has run out of available brokers to talk to: EOF

Sarama logs

``` 2025/05/13 18:26:29 Initializing new client 2025/05/13 18:26:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific. 2025/05/13 18:26:29 ClientID is the default of 'sarama', you should consider setting it to something application-specific. 2025/05/13 18:26:29 client/metadata fetching metadata for all topics from broker localhost:9092 2025/05/13 18:26:29 Connected to broker at localhost:9092 (unregistered) 2025/05/13 18:26:29 client/metadata got error from broker -1 while fetching metadata: EOF 2025/05/13 18:26:29 Closed connection to broker localhost:9092 2025/05/13 18:26:29 client/metadata no available broker to send metadata request to 2025/05/13 18:26:29 client/brokers resurrecting 1 dead seed brokers 2025/05/13 18:26:29 Error while sending ApiVersionsRequest to broker localhost:9092: kafka: broker not connected 2025/05/13 18:26:30 client/metadata retrying after 250ms... (2 attempts remaining) 2025/05/13 18:26:30 ClientID is the default of 'sarama', you should consider setting it to something application-specific. 2025/05/13 18:26:30 client/metadata fetching metadata for all topics from broker localhost:9092 2025/05/13 18:26:30 Connected to broker at localhost:9092 (unregistered) 2025/05/13 18:26:30 client/metadata got error from broker -1 while fetching metadata: EOF 2025/05/13 18:26:30 Closed connection to broker localhost:9092 2025/05/13 18:26:30 client/metadata no available broker to send metadata request to 2025/05/13 18:26:30 client/brokers resurrecting 1 dead seed brokers 2025/05/13 18:26:30 client/metadata retrying after 250ms... (1 attempts remaining) 2025/05/13 18:26:30 ClientID is the default of 'sarama', you should consider setting it to something application-specific. 2025/05/13 18:26:30 client/metadata fetching metadata for all topics from broker localhost:9092 2025/05/13 18:26:30 Connected to broker at localhost:9092 (unregistered) 2025/05/13 18:26:30 client/metadata got error from broker -1 while fetching metadata: EOF 2025/05/13 18:26:30 Closed connection to broker localhost:9092 2025/05/13 18:26:30 client/metadata no available broker to send metadata request to 2025/05/13 18:26:30 client/brokers resurrecting 1 dead seed brokers 2025/05/13 18:26:30 client/metadata retrying after 250ms... (0 attempts remaining) 2025/05/13 18:26:30 ClientID is the default of 'sarama', you should consider setting it to something application-specific. 2025/05/13 18:26:30 client/metadata fetching metadata for all topics from broker localhost:9092 2025/05/13 18:26:30 Connected to broker at localhost:9092 (unregistered) 2025/05/13 18:26:30 client/metadata got error from broker -1 while fetching metadata: EOF 2025/05/13 18:26:30 Closed connection to broker localhost:9092 2025/05/13 18:26:30 client/metadata no available broker to send metadata request to 2025/05/13 18:26:30 client/brokers resurrecting 1 dead seed brokers 2025/05/13 18:26:30 Closing Client ```

The culprit being simply that:

1. Sarama chooses request versions solely according to the provided config.Version, even though it sends an ApiVersionsRequest
2. MetadataRequest uses versions 8-9-10 for Kafka versions >= v2.4.0.0
3. Redpanda only supports versions 0 to 8, and it correctly advertises it

[puellanivis]

Nothing big.

[trapped]

@puellanivis I've addressed your review comments - would love it if you could have a second look when you've got some time! Thanks 🙂

[puellanivis]

Looks pretty good from my side. (note: I do not have permissions to approve PRs.)

[dnwe]

Thanks for working on this PR. Changes look good to me too so I’ll go ahead and merge into main. I’ll roll the main branch client out to a few places and leave it running for a bit before I cut a release

I’m also wondering about enabling usingApiVersions by default too and making DefaultVersion max too

[ijuma]

Great to see this, thanks!

[trapped]

After some deeper testing with Redpanda (I've found its small differences to be helpful in testing protocol compliance), I've noticed something that I think could use some further improvement in a new PR.

The good:

• in sendInternal(), we now use restrictApiVersion on the request's protocolBody; request payload encoding happens further down and respects rb.version()
• in handleResponsePromise, versionedDecode is called passing the request's message version, correctly matching request with response versions

The bad:

• in responseReceiver(), the response header version used is set ahead of time, when initializing the response promise
• however, when initializing the response promise the request's version hasn't been restricted yet!

In my testing I could reproduce this issue as follows:

1. setting config.Version = sarama.MaxVersion
2. pointing to Redpanda (latest version)
3. Sarama tries connecting to brokers, negotiating API versions; max version for MetadataRequest is advertised as 8, while the max supported by Sarama is 10
4. the response promise is initialized with header version 1, corresponding to MetadataResponse v10
5. the MetadataRequest v8 succeeds, and Redpanda returns a MetadataResponse v8 - header version 0 (flexibleVersions from v9 onwards for MetadataResponse, and we're using v8)
6. Sarama tries parsing the response payload, starting with the header: because Sarama thinks the header version is v1, it consumes 9 bytes instead of 8
7. Sarama parses the header successfully and proceeds to read the MetadataResponse body, but misaligned by 1 byte: instead of a 1-element brokers array, it thinks there's 256; instead of a 9-byte host string, it tries reading a 2403-byte string, which fails with "not enough bytes".

Replacing the response promise's headerVersion with a pointer to the response interface, which allows sendInternal() to update its version, could be a potential solution: #3223

Let me know what you think!

[ijuma]

The selected version is the highest supported by both the broker and the user-defined Config.Version upper bound.

Have we changed the default for the user-defined Config.Version upper bound? Ideally that would unset by default like most kafka client libraries (i.e. the client should use the highest version supported by the client and the broker).

[dnwe]

@ijuma yep that is not included in this PR, for staging purposes, but is the intention before cutting the next release. Currently trialling it via the FV in #3223

[ijuma]

Makes sense, thanks.