junrao commented on code in PR #23665:
URL: https://github.com/apache/kafka/pull/23665#discussion_r4232914829
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -153,8 +153,13 @@ public static RequestHeader parse(ByteBuffer buffer) {
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);
+ final RequestHeaderData headerData;
+ if (apiKey.isVersionSupported(apiVersion)) {
+ buffer.position(bufferStartPositionForHeader);
+ headerData = new RequestHeaderData(new
ByteBufferAccessor(buffer), headerVersion);
+ } else {
+ headerData = readCommonHeaderFields(buffer, apiKeyId,
apiVersion);
Review Comment:
This creates a slight behavior change. Since we skip reading the remaining
bytes in the header, requestContext.parseRequest() will read those bytes as
part of the request body. The result is indeterministic depending on whether
those bytes match the request body fields or not. We probably want to change
requestContext.parseRequest() to throw UnsupportedVersionException if
header.isApiVersionSupported() is false without parsing the body.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -177,6 +182,18 @@ public static RequestHeader parse(ByteBuffer buffer) {
}
}
+ private static RequestHeaderData readCommonHeaderFields(ByteBuffer buffer,
short apiKeyId, short apiVersion) {
Review Comment:
This will parse the header with the wrong header version. It's probably
better to do new RequestHeaderData(new ByteBufferAccessor(buffer), 1).
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -153,8 +153,13 @@ public static RequestHeader parse(ByteBuffer buffer) {
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);
+ final RequestHeaderData headerData;
+ if (apiKey.isVersionSupported(apiVersion)) {
+ buffer.position(bufferStartPositionForHeader);
+ headerData = new RequestHeaderData(new
ByteBufferAccessor(buffer), headerVersion);
+ } else {
Review Comment:
With this change, we need to adjust the comment in line 144.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -153,8 +153,13 @@ public static RequestHeader parse(ByteBuffer buffer) {
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);
+ final RequestHeaderData headerData;
Review Comment:
1. Could we add the javadoc for the method? We want to document how it
treats unsupported ApiVersionRequest.
2. Let's document in RequestHeader.json that we can only add tagged fields
until all supported releases have this change.
--
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]
