Jun Rao created KAFKA-21160:
-------------------------------
Summary: improve RequestHeader parsing on the broker
Key: KAFKA-21160
URL: https://issues.apache.org/jira/browse/KAFKA-21160
Project: Kafka
Issue Type: Improvement
Reporter: Jun Rao
During the discussion of KIP-1313, we realize that the current parsing logic on
the broker prevents us from adding a new non-tagged field in the request header
in the future. RequestHeader.parse() has the following code.
{code:java}
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);
{code}
Suppose that we bump up the request header version from 2 to 3 by adding a
non-tagged field in the future. When a new client tries to connect to an older
broker, it will send a new version of ApiVersionRequest including the new
header on a new connection. Since the old broker doesn't understand the new
version of the ApiVersionRequest, it sets headerVersion to 2. "new
RequestHeaderData" will use v2 header schema to parse the bytes serialized for
v3 request header. This will hit an InvalidRequestException, which will trigger
the closing of the connection. When the client reconnects, it will hit an
InvalidRequestException again when sending the new version of the
ApiVersionRequest. This means that a new client can never talk to an old broker.
One solution is to change RequestHeader.parse() to recognize unsupported
request by parsing just the first two fields apiKey and apiVersion in the
header. If apiVersion is not supported, we can generate a RequestHeader with an
unsupported flag without parsing the rest of the bytes. This way, the request
header will be propagated to RequestContext.parseRequest() and a v0
ApiVersionResponse will be sent to the client for it to make progress. Once all
old versions of the broker are phased out, we can start adding new non-tagged
fields to the RequestHeader, as longer as we never change the first two fields.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
