junrao commented on code in PR #23665:
URL: https://github.com/apache/kafka/pull/23665#discussion_r4339262489
##########
clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java:
##########
@@ -96,7 +96,11 @@ public AbstractRequest(ApiKeys apiKey, short version) {
}
/**
- * Get the version of this AbstractRequest object.
+ * Returns the API version of this request object, which determines its
body format.
Review Comment:
AbstractRequest is a generic class and not used just by request handlers.
version() should be self explanatory, i.e., it's the version associated with
this request object. We can just remove the comment here.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -148,15 +171,32 @@ public Send buildResponseSend(AbstractResponse body) {
* so we do not lose the benefit of "zero copy" transfers from disk.
*/
public ByteBuffer buildResponseEnvelopePayload(AbstractResponse body) {
- return body.serializeWithHeader(header.toResponseHeader(),
apiVersion());
+ short version = effectiveApiVersion();
+ return body.serializeWithHeader(responseHeader(version), version);
+ }
+
+ /**
+ * The response header follows the version used to serialize the response
body. For an unsupported
+ * ApiVersionsRequest, this is v0 rather than the original request version.
Review Comment:
This is a generic method and is self explanatory. We can remove the comment.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -191,6 +231,13 @@ public int requestType() {
return header.apiKey().id;
}
+ /**
+ * Returns the original API version sent by the client in the request
header. This can differ from
+ * {@link #effectiveApiVersion()} for an unsupported {@link
ApiVersionsRequest}. Use this when the original
+ * version is needed, such as checking whether the requested API version
is enabled.
+ *
+ * @see RequestHeader#apiVersion()
Review Comment:
We can remove this line.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -63,6 +63,13 @@ public ApiKeys apiKey() {
return ApiKeys.forId(data.requestApiKey());
}
+ /**
+ * Returns the API version stored in the request header, even if the
server does not support it.
+ * This is the original version sent by the client, not necessarily the
version used to process the request.
+ * Server-side request handlers should use {@link
RequestContext#effectiveApiVersion()} or
+ * {@link AbstractRequest#version()} for version-dependent behavior, and
{@link RequestContext#requestVersion()}
+ * when the original client version is needed.
+ */
Review Comment:
The API should be self explanatory. The server side details are explained in
RequestContext already. We can remove this comment.
We can probably document RequestContext.header that this is the original
header from the client request. For unsupported ApiVerionRequest, use
effectiveApiVersion() to get the actual parsed API version.
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1058,7 +1059,7 @@ class KafkaApis(val requestChannel: RequestChannel,
CompletableFuture.allOf(futures.toArray: _*).handle[Unit] { (_, _) =>
val groupResponses = new
ArrayBuffer[OffsetFetchResponseData.OffsetFetchResponseGroup](futures.size)
futures.foreach(future => groupResponses += future.get())
- requestHelper.sendMaybeThrottle(request, new
OffsetFetchResponse.Builder(groupResponses.asJava).build(request.context.apiVersion))
+ requestHelper.sendMaybeThrottle(request, new
OffsetFetchResponse.Builder(groupResponses.asJava).build(request.context.effectiveApiVersion))
Review Comment:
Probably use offsetFetchRequest.version here?
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -148,15 +171,32 @@ public Send buildResponseSend(AbstractResponse body) {
* so we do not lose the benefit of "zero copy" transfers from disk.
*/
public ByteBuffer buildResponseEnvelopePayload(AbstractResponse body) {
- return body.serializeWithHeader(header.toResponseHeader(),
apiVersion());
+ short version = effectiveApiVersion();
+ return body.serializeWithHeader(responseHeader(version), version);
+ }
+
+ /**
+ * The response header follows the version used to serialize the response
body. For an unsupported
+ * ApiVersionsRequest, this is v0 rather than the original request version.
+ */
+ private ResponseHeader responseHeader(short responseVersion) {
+ return new ResponseHeader(header.correlationId(),
header.apiKey().responseHeaderVersion(responseVersion));
}
private boolean isUnsupportedApiVersionsRequest() {
return header.apiKey() == API_VERSIONS &&
!header.isApiVersionSupported();
}
- public short apiVersion() {
- // Use v0 when serializing an unhandled ApiVersion response
+ /**
+ * Returns the API version used to parse the request body and serialize
the response.
+ * An unsupported {@link ApiVersionsRequest} is treated as v0 without
parsing its body, and its response
+ * is serialized as v0. For all other requests, this is the original
version sent by the client.
+ * Request handlers should use this version or {@link
AbstractRequest#version()} for version-dependent behavior.
+ *
+ * @see #requestVersion()
+ * @see AbstractRequest#version()
Review Comment:
How about the following?
Returns the API version parsed by the server. This is usually the API
version in the original client request. The only exception is an unsupported
{@link ApiVersionsRequest}, whose API version is set to 0 after parsing. The
response should be sent and serialized using this version.
To retrieve the API version in the original client request, use
#requestVersion() instead.
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -191,6 +231,13 @@ public int requestType() {
return header.apiKey().id;
}
+ /**
+ * Returns the original API version sent by the client in the request
header. This can differ from
Review Comment:
It seems that all usage of requestVersion() in GroupCoordinatorService,
GroupMetadataManager and OffsetMetadataManager should be using
effectiveApiVersion.
--
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]
