chia7712 commented on code in PR #23665:
URL: https://github.com/apache/kafka/pull/23665#discussion_r4344793176
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestContext.java:
##########
@@ -148,15 +175,26 @@ 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);
+ }
+
+ private ResponseHeader responseHeader(short responseVersion) {
+ return new ResponseHeader(header.correlationId(),
header.apiKey().responseHeaderVersion(responseVersion));
}
private boolean isUnsupportedApiVersionsRequest() {
Review Comment:
Should we consolidate the fallback logic into ApiKeys? For example:
```java
public enum ApiKeys {
...
/**
* ApiVersions is used for version discovery, so the server accepts any
version of it.
* An unsupported version is handled as this version, and the client
falls back to parsing
* the response with it (KIP-511). This version must never be removed.
*/
public static final short API_VERSIONS_FALLBACK_VERSION = 0;
/**
* Returns true if a request with the given version must be handled as
* {@link #API_VERSIONS_FALLBACK_VERSION} instead of being parsed.
*/
public boolean isFallbackVersion(short version) {
return this == API_VERSIONS && !isVersionSupported(version);
}
public boolean isVersionEnabled(short apiVersion, boolean
enableUnstableLastVersion) {
// ApiVersions accepts any version, see isFallbackVersion
if (this == API_VERSIONS) return true;
...
}
}
```
--
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]
