junrao commented on code in PR #23665:
URL: https://github.com/apache/kafka/pull/23665#discussion_r4252952021
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -137,11 +137,28 @@ public ResponseHeader toResponseHeader() {
return new ResponseHeader(data.correlationId(),
apiKey().responseHeaderVersion(apiVersion()));
}
+ /**
+ * Parses a request header from the given buffer. On return, the buffer is
positioned at the start of the
Review Comment:
the buffer is positioned => for a supported request API version, the buffer
is positioned
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -137,11 +137,28 @@ public ResponseHeader toResponseHeader() {
return new ResponseHeader(data.correlationId(),
apiKey().responseHeaderVersion(apiVersion()));
}
+ /**
+ * Parses a request header from the given buffer. On return, the buffer is
positioned at the start of the
+ * request body.
+ * <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.
+ * This allows the broker to return an error response for requests with
unsupported versions, particularly
+ * {@link ApiVersionsRequest} from newer clients. The request body must
not be parsed in this case because the buffer
Review Comment:
The broker actually doesn't send unsupported versions for unsupported
ApiVersionsRequest. It generates a V0 ApiVersionsResponse instead. The client
will parse the response as the expected version and on parsing exception, fall
to parsing it as V0.
##########
clients/src/main/resources/common/message/RequestHeader.json:
##########
@@ -40,6 +40,11 @@
// 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.
+ //
+ // When the request api version is unsupported, the request header is
parsed with version 1,
+ // i.e. only the fields above are read (see RequestHeader#parse). Brokers
without that change
+ // parse the header with the version derived from the request api version
instead. So, any new
+ // field added to the header must be a tagged field until all supported
releases have that change.
Review Comment:
How about the following?
Currently, we can't add new non-tagged fields to the request header for
ApiVersionRequest. This is because the old releases have a bug on the broker
that parses unknown request header with the first 4 fields and the tagged
fields. If a new client sends an ApiVersionRequest with a new non-tagged field
in the header to an old broker, the older 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 support
API versions on the old broker. We have since fixed the broker to parse
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 are preserved. However, we need to wait until all
supported releases have that fix. See KAFKA-21160 for more details.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -153,8 +170,9 @@ public static RequestHeader parse(ByteBuffer buffer) {
throw new InvalidRequestException("Unsupported api with key "
+ apiKeyId + " (" + apiKey.name + ") and version " + apiVersion);
short headerVersion = apiKey.requestHeaderVersion(apiVersion);
+ short parseHeaderVersion = apiKey.isVersionSupported(apiVersion) ?
headerVersion : (short) 1;
buffer.position(bufferStartPositionForHeader);
- final RequestHeaderData headerData = new RequestHeaderData(new
ByteBufferAccessor(buffer), headerVersion);
+ final RequestHeaderData headerData = new RequestHeaderData(new
ByteBufferAccessor(buffer), parseHeaderVersion);
Review Comment:
We should use parseHeaderVersion when constructing RequestHeader below,
right?
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -137,11 +137,28 @@ public ResponseHeader toResponseHeader() {
return new ResponseHeader(data.correlationId(),
apiKey().responseHeaderVersion(apiVersion()));
}
+ /**
+ * Parses a request header from the given buffer. On return, the buffer is
positioned at the start of the
+ * request body.
+ * <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.
Review Comment:
We probably should add in that case, the returned header will have false in
header.isApiVersionSupported() and the caller is expected to check that before
parsing the rest of the bytes in the request.
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java:
##########
@@ -104,6 +106,74 @@ public void testClientInstanceIdIsUnsetBelowTheV3Header() {
assertEquals(header, RequestHeader.parse(buffer));
}
+ @Test
+ public void testParseUnsupportedApiVersionOnlyReadsCommonHeaderFields() {
+ short apiVersion = (short) (ApiKeys.API_VERSIONS.latestVersion() + 1);
+ byte[] unknownField = unknownHeaderField();
+ ByteBuffer buffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.API_VERSIONS, apiVersion,
123,
+ "client", unknownField);
+ int commonFieldsSize = buffer.remaining() - unknownField.length;
+
+ RequestHeader header = RequestHeader.parse(buffer);
+ assertEquals(ApiKeys.API_VERSIONS, header.apiKey());
+ assertEquals(apiVersion, header.apiVersion());
+ assertEquals(123, header.correlationId());
+ assertEquals("client", header.clientId());
+ assertFalse(header.isApiVersionSupported());
+ assertEquals(ApiKeys.API_VERSIONS.requestHeaderVersion(apiVersion),
header.headerVersion());
+ // Only the common fields are consumed and the unknown bytes are left
in the buffer.
+ assertEquals(commonFieldsSize, header.size());
+ assertEquals(commonFieldsSize, buffer.position());
+ assertEquals(unknownField.length, buffer.remaining());
+ }
+
+ @Test
+ public void testParseSupportedApiVersionRejectsInvalidTaggedFields() {
+ ByteBuffer buffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.API_VERSIONS,
+ ApiKeys.API_VERSIONS.latestVersion(), 123, "client",
unknownHeaderField());
+ assertThrows(InvalidRequestException.class, () ->
RequestHeader.parse(buffer));
+ }
+
+ @Test
+ public void testParseUnsupportedApiVersionWithNullClientId() {
+ ByteBuffer buffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.API_VERSIONS,
Short.MAX_VALUE, 123,
+ null, new byte[0]);
+ RequestHeader header = RequestHeader.parse(buffer);
+ assertEquals("", header.clientId());
Review Comment:
Perhaps add a comment why the clientId is empty instead of null.
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestContextTest.java:
##########
@@ -63,6 +65,34 @@ public void testSerdeUnsupportedApiVersionRequest() throws
Exception {
requestBuffer.putInt(29034);
requestBuffer.flip();
+ verifyUnsupportedApiVersionsRequestSerde(context, requestBuffer,
correlationId);
+ }
+
+ @Test
+ public void testSerdeUnsupportedApiVersionRequestWithUnknownHeaderFields()
throws Exception {
+ int correlationId = 23423;
+ short apiVersion = (short) (ApiKeys.API_VERSIONS.latestVersion() + 1);
+
+ // A newer client may use a header version unknown to this broker,
with fields after the client id which
+ // cannot be parsed. The header parser only reads the fields common to
every header version, and the rest
+ // of the buffer is ignored since the request is treated as v0.
+ byte[] unknownBytes = new byte[24];
+ Arrays.fill(unknownBytes, (byte) 0xFF);
+ ByteBuffer requestBuffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.API_VERSIONS, apiVersion,
+ correlationId, "client", unknownBytes);
Review Comment:
This is an existing issue, but we can just do `request.getErrorResponse(0,
Errors.UNSUPPORTED_VERSION.exception())`.
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java:
##########
@@ -104,6 +106,74 @@ public void testClientInstanceIdIsUnsetBelowTheV3Header() {
assertEquals(header, RequestHeader.parse(buffer));
}
+ @Test
+ public void testParseUnsupportedApiVersionOnlyReadsCommonHeaderFields() {
+ short apiVersion = (short) (ApiKeys.API_VERSIONS.latestVersion() + 1);
+ byte[] unknownField = unknownHeaderField();
+ ByteBuffer buffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.API_VERSIONS, apiVersion,
123,
+ "client", unknownField);
+ int commonFieldsSize = buffer.remaining() - unknownField.length;
+
+ RequestHeader header = RequestHeader.parse(buffer);
+ assertEquals(ApiKeys.API_VERSIONS, header.apiKey());
+ assertEquals(apiVersion, header.apiVersion());
+ assertEquals(123, header.correlationId());
+ assertEquals("client", header.clientId());
+ assertFalse(header.isApiVersionSupported());
+ assertEquals(ApiKeys.API_VERSIONS.requestHeaderVersion(apiVersion),
header.headerVersion());
+ // Only the common fields are consumed and the unknown bytes are left
in the buffer.
+ assertEquals(commonFieldsSize, header.size());
+ assertEquals(commonFieldsSize, buffer.position());
+ assertEquals(unknownField.length, buffer.remaining());
+ }
+
+ @Test
+ public void testParseSupportedApiVersionRejectsInvalidTaggedFields() {
+ ByteBuffer buffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.API_VERSIONS,
+ ApiKeys.API_VERSIONS.latestVersion(), 123, "client",
unknownHeaderField());
+ assertThrows(InvalidRequestException.class, () ->
RequestHeader.parse(buffer));
+ }
+
+ @Test
+ public void testParseUnsupportedApiVersionWithNullClientId() {
+ ByteBuffer buffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.API_VERSIONS,
Short.MAX_VALUE, 123,
+ null, new byte[0]);
+ RequestHeader header = RequestHeader.parse(buffer);
+ assertEquals("", header.clientId());
+ assertEquals(123, header.correlationId());
+ assertFalse(header.isApiVersionSupported());
+ assertEquals(0, buffer.remaining());
+ }
+
+ @Test
+ public void testParseUnsupportedVersionOfOtherApi() {
+ ByteBuffer buffer =
RequestTestUtils.serializeRequestHeaderPrefix(ApiKeys.FETCH, Short.MAX_VALUE,
5, "client",
+ unknownHeaderField());
+ RequestHeader header = RequestHeader.parse(buffer);
+ assertEquals(ApiKeys.FETCH, header.apiKey());
+ assertEquals(Short.MAX_VALUE, header.apiVersion());
+ assertEquals(5, header.correlationId());
+ assertEquals("client", header.clientId());
+ assertFalse(header.isApiVersionSupported());
Review Comment:
Should we assert buffer.position() and remaining(), as the ApiVersions test
does?
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -114,6 +115,12 @@ public RequestAndSize parseRequest(ByteBuffer buffer) {
// Unsupported ApiVersion requests are treated as v0 requests and
are not parsed
ApiVersionsRequest apiVersionsRequest = new ApiVersionsRequest(new
ApiVersionsRequestData(), (short) 0, header.apiVersion());
return new RequestAndSize(apiVersionsRequest, 0);
+ } else if (!header.isApiVersionSupported()) {
Review Comment:
Let's document the exceptions in the javadoc of the method. Since it now
also throws UnsupportedVersionException, it's better to handle it explicitly in
the callers SaslServerAuthenticator.handleKafkaRequest() and
EnvelopUtils.parseForwardedRequest() so that we can get the same error message.
Also, the isApiVersionSupported() check in
SaslServerAuthenticator.handleSaslToken (:454-458) is already unreachable and
could be removed.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -153,8 +170,9 @@ public static RequestHeader parse(ByteBuffer buffer) {
throw new InvalidRequestException("Unsupported api with key "
+ apiKeyId + " (" + apiKey.name + ") and version " + apiVersion);
short headerVersion = apiKey.requestHeaderVersion(apiVersion);
+ short parseHeaderVersion = apiKey.isVersionSupported(apiVersion) ?
headerVersion : (short) 1;
Review Comment:
Would it be better to just do `short headerVersion =
apiKey.isVersionSupported(apiVersion) ? apiKey.requestHeaderVersion(apiVersion)
: (short) 1;`?
--
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]
