junrao commented on code in PR #23665:
URL: https://github.com/apache/kafka/pull/23665#discussion_r4275776172
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -109,11 +110,26 @@ public RequestContext(RequestHeader header,
this.principalSerde = principalSerde;
}
+ /**
+ * Parses the request body from the given buffer, which must be positioned
at the start of the request body.
+ * An {@link ApiVersionsRequest} with an unsupported version is not parsed
and is treated as a v0 request.
Review Comment:
> An {@link ApiVersionsRequest} with an unsupported version is not parsed
and is treated as a v0 request.
>
If the header indicates the request version is unsupported, the request body
is not parsed. An unsupported {@link ApiVersionsRequest} is treated as a v0
request. For other requests, an UnsupportedVersionException is thrown.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -137,11 +137,31 @@ public ResponseHeader toResponseHeader() {
return new ResponseHeader(data.correlationId(),
apiKey().responseHeaderVersion(apiVersion()));
}
+ /**
+ * Parses a request header from the given buffer. For a supported request
API version, the buffer is positioned
+ * at the start of the request body on return.
+ * <p>
+ * If the request API version is not supported by this node, the
corresponding header version may also be unknown.
+ * In this case, the header is parsed using version 1, which contains only
the fields common to all header versions:
+ * API key, API version, correlation ID, and client ID. Any remaining
header bytes, including tagged fields,
+ * are left unread. The returned header has {@link
#isApiVersionSupported()} set to false, and the caller is
+ * expected to check it before parsing the rest of the bytes in the
request, since the buffer is not positioned
+ * at the start of the request body.
+ * <p>
+ * This allows the broker to handle {@link ApiVersionsRequest} from newer
clients, which uses a request version
+ * unknown to the broker. The broker responds with a v0 {@link
ApiVersionsResponse} in that case, and the client
+ * falls back to parsing the response as v0 if it cannot parse it as the
expected version.
Review Comment:
This comment seems to belong to RequestContext.parseRequest.
##########
clients/src/main/resources/common/message/RequestHeader.json:
##########
@@ -40,6 +40,17 @@
// 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.
+ //
+ // Currently, we can't add new non-tagged fields to the request header for
ApiVersionsRequest.
+ // This is because the old releases have a bug on the broker that parses
an unknown request header
+ // with the first 4 fields and the tagged fields. If a new client sends an
ApiVersionsRequest with a
+ // new non-tagged field in the header to an old broker, the old broker
will fail parsing the new field
+ // as the tagged fields and will throw an InvalidRequestException instead
of sending a v0 response.
+ // Therefore, the new client can never discover the supported API versions
on the old broker.
+ // We have since fixed the broker to parse an unsupported request header
with only the first 4 fields.
Review Comment:
We have since fixed the broker => In 4.5.0, we have fixed the broker
##########
clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java:
##########
@@ -542,7 +537,7 @@ private void handleKafkaRequest(byte[] requestBytes) throws
IOException, Authent
setSaslState(SaslState.AUTHENTICATE);
}
}
- } catch (InvalidRequestException e) {
+ } catch (InvalidRequestException | UnsupportedVersionException e) {
Review Comment:
In the comment below, we want to say InvalidRequestException or
UnsupportedVersionException.
##########
clients/src/main/resources/common/message/RequestHeader.json:
##########
@@ -40,6 +40,17 @@
// 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.
+ //
+ // Currently, we can't add new non-tagged fields to the request header for
ApiVersionsRequest.
+ // This is because the old releases have a bug on the broker that parses
an unknown request header
+ // with the first 4 fields and the tagged fields. If a new client sends an
ApiVersionsRequest with a
+ // new non-tagged field in the header to an old broker, the old broker
will fail parsing the new field
+ // as the tagged fields and will throw an InvalidRequestException instead
of sending a v0 response.
+ // Therefore, the new client can never discover the supported API versions
on the old broker.
+ // We have since fixed the broker to parse an unsupported request header
with only the first 4 fields.
+ // This will allow us to add non-tagged fields to the request header in
the future, as long as the
+ // order of the first 4 fields is preserved. However, we need to wait
until all supported releases
+ // have that fix. See KAFKA-21160 for more details.
Review Comment:
> we need to wait until all supported releases have that fix.
we need to wait until all releases before 4.5.0 are no longer supported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
