AndrewJSchofield commented on code in PR #23317:
URL: https://github.com/apache/kafka/pull/23317#discussion_r3950272566
##########
generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java:
##########
@@ -35,6 +36,8 @@ public final class ApiMessageTypeGenerator implements
TypeClassGenerator {
private final CodeBuffer buffer;
private final TreeMap<Short, ApiData> apis;
private final EnumMap<RequestListenerType, List<ApiData>> apisByListener =
new EnumMap<>(RequestListenerType.class);
+ private MessageSpec requestHeaderSpec;
Review Comment:
I wonder why these are not logically contained within the `requestSpec` and
`responseSpec` of the `ApiData` itself.
##########
generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java:
##########
@@ -141,6 +152,7 @@ public void generateAndWrite(BufferedWriter writer) throws
IOException {
}
private void generate() {
+ validateHeaderVersions();
Review Comment:
This validation code should not be called in the generation phase.
##########
clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java:
##########
@@ -98,6 +102,64 @@ public void testHeaderVersion() {
assertEquals((short) 2,
ApiMessageType.CREATE_TOPICS.requestHeaderVersion((short) 5));
assertEquals((short) 1,
ApiMessageType.CREATE_TOPICS.responseHeaderVersion((short) 5));
+
+ // SaslHandshake and OffsetDelete are non-flexible: header v1 request
/ v0 response at every version.
+ assertEquals((short) 1,
ApiMessageType.SASL_HANDSHAKE.requestHeaderVersion((short) 0));
+ assertEquals((short) 0,
ApiMessageType.SASL_HANDSHAKE.responseHeaderVersion((short) 0));
+ assertEquals((short) 1,
ApiMessageType.SASL_HANDSHAKE.requestHeaderVersion((short) 1));
+ assertEquals((short) 0,
ApiMessageType.SASL_HANDSHAKE.responseHeaderVersion((short) 1));
+
+ assertEquals((short) 1,
ApiMessageType.OFFSET_DELETE.requestHeaderVersion((short) 0));
+ assertEquals((short) 0,
ApiMessageType.OFFSET_DELETE.responseHeaderVersion((short) 0));
+
+ // ApiVersions request follows the flexible rule, but the response
always uses a v0 header (KIP-511).
+ assertEquals((short) 1,
ApiMessageType.API_VERSIONS.requestHeaderVersion((short) 0));
+ assertEquals((short) 1,
ApiMessageType.API_VERSIONS.requestHeaderVersion((short) 2));
+ assertEquals((short) 2,
ApiMessageType.API_VERSIONS.requestHeaderVersion((short) 3));
+ assertEquals((short) 0,
ApiMessageType.API_VERSIONS.responseHeaderVersion((short) 0));
+ assertEquals((short) 0,
ApiMessageType.API_VERSIONS.responseHeaderVersion((short) 3));
+
+ // Envelope is flexible from v0: header v2 request / v1 response
everywhere.
+ assertEquals((short) 2,
ApiMessageType.ENVELOPE.requestHeaderVersion((short) 0));
+ assertEquals((short) 1,
ApiMessageType.ENVELOPE.responseHeaderVersion((short) 0));
+ }
+
+ /**
+ * The header versions generated from the headerVersions maps match the
versions implied by body
+ * flexibility for every existing API and version: a flexible
request/response uses header v2/v1,
+ * a non-flexible one uses header v1/v0. The sole exception is
ApiVersionsResponse, which always
+ * uses a v0 header so that older brokers can parse the response header
(KIP-511).
+ */
+ @Test
+ public void testHeaderVersionsMatchSchemaFlexibility() {
+ for (ApiMessageType type : ApiMessageType.values()) {
+ if (type.lowestSupportedVersion() >
type.highestSupportedVersion(true))
+ continue;
+ for (short version = type.lowestSupportedVersion();
+ version <= type.highestSupportedVersion(true); version++) {
+ short expectedRequestHeader =
isFlexible(type.requestSchemas()[version]) ? (short) 2 : (short) 1;
+ assertEquals(expectedRequestHeader,
type.requestHeaderVersion(version),
Review Comment:
This is going to age badly. When you add v3 request headers, then the
request header implied by `isFlexible` is no longer sufficient. I suppose that
you can assert `>=`. I'm still contemplating how to test RPC header consistency
across AK releases so we don't inadvertently make new code use a higher header
version than expected.
##########
clients/src/main/resources/common/message/AlterPartitionRequest.json:
##########
@@ -26,6 +26,9 @@
// Version 3 adds the NewIsrEpochs field and deprecates the NewIsr field
(KIP-903).
"validVersions": "2-3",
"flexibleVersions": "0+",
+ "headerVersions": {
Review Comment:
In a few cases, we have RPCs in which the oldest versions are no longer
valid versions. The JSON schemas still describe all of the versions, even
though the most recent brokers no longer support them. That's why you see
`validVersions: 2-3` but `flexibleVersions: 0+`. My view is that the header
versions ought to cover the entire range too, so I would expect `"0+": "2"`
here. There are some other RPCs, such as `ShareFetch` which this would also
affect.
##########
clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java:
##########
@@ -234,7 +234,9 @@ private String apiVersionToText(ApiVersion apiVersion) {
bld.append(" [unusable: node too old]");
} else {
short latestUsableVersion = Utils.min(apiKey.latestVersion(),
supportedVersion.maxVersion());
- bld.append(" [usable:
").append(latestUsableVersion).append("]");
+ bld.append(" [usable: ").append(latestUsableVersion)
+ .append(", request header:
").append(apiKey.requestHeaderVersion(latestUsableVersion))
Review Comment:
I think this change should be reverted. It is nice to be able to see the
effect of the PR interactively, but it was not mentioned in the KIP and I was
not anticipating this change to these command-line tools. The revised output
doesn't quite seem that useful to me in practice because it's just the request
header version of the latest usable RPC version, and the real picture is much
more complicated (and not really something that I think we should put in the
output of the tools).
--
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]
