You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dajac (via GitHub)" <gi...@apache.org> on 2023/07/19 12:24:28 UTC
[GitHub] [kafka] dajac opened a new pull request, #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
dajac opened a new pull request, #14046:
URL: https://github.com/apache/kafka/pull/14046
This patch does a few things:
1) It introduces version 9 of the OffsetCommit API. This new version has no schema changes but it can return a StaleMemberEpochException if the new consumer group protocol is used.
2) It renames the `generationId` field in the request to `GenerationIdOrMemberEpoch`. This is backward compatible change.
3) It introduces the new StaleMemberEpochException error.
4) It does a minor refactoring in OffsetCommitRequest class.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #14046:
URL: https://github.com/apache/kafka/pull/14046
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1644292908
Looking at the tests `[Build / JDK 20 and Scala 2.13 / kafka.server.FetchRequestTest.testCurrentEpochValidationV12()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-14046/7/testReport/junit/kafka.server/FetchRequestTest/Build___JDK_20_and_Scala_2_13___testCurrentEpochValidationV12__/)` is a bit strange but it only failed on that version. Everything else seems to be familiar-ish flakes
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14046:
URL: https://github.com/apache/kafka/pull/14046#discussion_r1270401089
##########
clients/src/main/resources/common/message/OffsetCommitRequest.json:
##########
@@ -31,13 +31,19 @@
// version 7 adds a new field called groupInstanceId to indicate member identity across restarts.
//
// Version 8 is the first flexible version.
- "validVersions": "0-8",
+ //
+ // Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). The
+ // request is the same as version 8.
+ // Version 9 is added as part of KIP-848 and is still under development. Hence, the last version of the
+ // API is not exposed by default by brokers unless explicitly enabled.
+ "latestVersionUnstable": true,
+ "validVersions": "0-9",
"flexibleVersions": "8+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The unique group identifier." },
- { "name": "GenerationId", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
- "about": "The generation of the group." },
+ { "name": "GenerationIdOrMemberEpoch", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
+ "about": "The generation of the group if the generic group protocol or the member epoch if the consumer protocol." },
Review Comment:
> is it safe to change the field name?
Yes, it is as this does not change the serialized bytes.
> you're saying based on the consumer since we can have a group with both new & old consumers right?
It is really based on the group type. If we have a group with both old and new consumers, it means that the group is being upgraded so the new "engine" is used for both in this case and the member epoch is the expected value for both as well. Of course, the old consumers won't really know about this. They will believe that they send the generation id.
> nit: "if using ... if using ..."
Ack.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a diff in pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "mumrah (via GitHub)" <gi...@apache.org>.
mumrah commented on code in PR #14046:
URL: https://github.com/apache/kafka/pull/14046#discussion_r1269710067
##########
clients/src/main/java/org/apache/kafka/common/errors/StaleMemberEpochException.java:
##########
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+import org.apache.kafka.common.annotation.InterfaceStability;
+
+@InterfaceStability.Evolving
+public class StaleMemberEpochException extends ApiException {
Review Comment:
I know there isn't much precedent for this, but it might be useful to include a doc string here explaining which RPC this error is used in and at what version
##########
clients/src/main/resources/common/message/OffsetCommitRequest.json:
##########
@@ -31,13 +31,19 @@
// version 7 adds a new field called groupInstanceId to indicate member identity across restarts.
//
// Version 8 is the first flexible version.
- "validVersions": "0-8",
+ //
+ // Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). The
+ // request is the same as version 8.
+ // Version 9 is added as part of KIP-848 and is still under development. Hence, the last version of the
+ // API is not exposed by default by brokers unless explicitly enabled.
+ "latestVersionUnstable": true,
+ "validVersions": "0-9",
"flexibleVersions": "8+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The unique group identifier." },
- { "name": "GenerationId", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
- "about": "The generation of the group." },
+ { "name": "GenerationIdOrMemberEpoch", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
+ "about": "The generation of the group if the generic group protocol or the member epoch if the consumer protocol." },
Review Comment:
How does the server decide to interpret this value as a GenerationId vs a MemberEpoch? Is it based on the API version used?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1644264091
> @jolshan I was actually thinking about the AuthorizerIntegrationTest failures overnight and I found an issue with the latestVersionUnstable flag. Let me try to explain.
I was curious if the unstable version flag was causing issues since I recall some weirdness in tests when I had an unstable version. Makes sense to require the unstable-ness to be explicit, but I will take a second look.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1646067746
Failed tests are unrelated. Merging to trunk.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1642653815
> Something strange is going on with AuthorizerIntegrationTest in the build, but that might be unrelated. I will look into that.
@jolshan I found the issue related to AuthorizerIntegrationTest failures. I just pushed a fix. See last commit.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1642696317
Thanks for the fix @dajac let's just confirm the build and we should be good :)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on a diff in pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #14046:
URL: https://github.com/apache/kafka/pull/14046#discussion_r1269711912
##########
clients/src/main/resources/common/message/OffsetCommitRequest.json:
##########
@@ -31,13 +31,19 @@
// version 7 adds a new field called groupInstanceId to indicate member identity across restarts.
//
// Version 8 is the first flexible version.
- "validVersions": "0-8",
+ //
+ // Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). The
+ // request is the same as version 8.
+ // Version 9 is added as part of KIP-848 and is still under development. Hence, the last version of the
+ // API is not exposed by default by brokers unless explicitly enabled.
+ "latestVersionUnstable": true,
+ "validVersions": "0-9",
"flexibleVersions": "8+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The unique group identifier." },
- { "name": "GenerationId", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
- "about": "The generation of the group." },
+ { "name": "GenerationIdOrMemberEpoch", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
+ "about": "The generation of the group if the generic group protocol or the member epoch if the consumer protocol." },
Review Comment:
The new group coordinator uses the member epoch and the old one uses the generation id I believe.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1642437560
Something strange is going on with AuthorizerIntegrationTest in the build, but that might be unrelated. I will look into that.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14046:
URL: https://github.com/apache/kafka/pull/14046#discussion_r1268592972
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java:
##########
@@ -89,33 +88,29 @@ public Map<TopicPartition, Long> offsets() {
return offsets;
}
- public static List<OffsetCommitResponseTopic> getErrorResponseTopics(
- List<OffsetCommitRequestTopic> requestTopics,
Review Comment:
The usage below was the single usage of this method so we always used it with all topics. For the context, I refactored it because I use the refactored one in one of the next patch.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1642424251
> Note the use of "latestVersionUnstable": true in the request schema. This means that this new version is not available yet unless activated.
This also leaves us room to add the topic ID to this version if we so choose?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a diff in pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14046:
URL: https://github.com/apache/kafka/pull/14046#discussion_r1269732280
##########
clients/src/main/resources/common/message/OffsetCommitRequest.json:
##########
@@ -31,13 +31,19 @@
// version 7 adds a new field called groupInstanceId to indicate member identity across restarts.
//
// Version 8 is the first flexible version.
- "validVersions": "0-8",
+ //
+ // Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). The
+ // request is the same as version 8.
+ // Version 9 is added as part of KIP-848 and is still under development. Hence, the last version of the
+ // API is not exposed by default by brokers unless explicitly enabled.
+ "latestVersionUnstable": true,
+ "validVersions": "0-9",
"flexibleVersions": "8+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The unique group identifier." },
- { "name": "GenerationId", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
- "about": "The generation of the group." },
+ { "name": "GenerationIdOrMemberEpoch", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
+ "about": "The generation of the group if the generic group protocol or the member epoch if the consumer protocol." },
Review Comment:
It based on the type of the group. In the new group coordinator, we have two types: generic (the old protocol) and consumer (the new protocol).
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jeffkbkim commented on a diff in pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #14046:
URL: https://github.com/apache/kafka/pull/14046#discussion_r1270133576
##########
clients/src/main/resources/common/message/OffsetCommitRequest.json:
##########
@@ -31,13 +31,19 @@
// version 7 adds a new field called groupInstanceId to indicate member identity across restarts.
//
// Version 8 is the first flexible version.
- "validVersions": "0-8",
+ //
+ // Version 9 is the first version that can be used with the new consumer group protocol (KIP-848). The
+ // request is the same as version 8.
+ // Version 9 is added as part of KIP-848 and is still under development. Hence, the last version of the
+ // API is not exposed by default by brokers unless explicitly enabled.
+ "latestVersionUnstable": true,
+ "validVersions": "0-9",
"flexibleVersions": "8+",
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The unique group identifier." },
- { "name": "GenerationId", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
- "about": "The generation of the group." },
+ { "name": "GenerationIdOrMemberEpoch", "type": "int32", "versions": "1+", "default": "-1", "ignorable": true,
+ "about": "The generation of the group if the generic group protocol or the member epoch if the consumer protocol." },
Review Comment:
3 comments:
* is it safe to change the field name?
* > It based on the type of the group
you're saying based on the consumer since we can have a group with both new & old consumers right?
* nit: "if using ... if using ..."
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1643522968
@jolshan I was actually thinking about the `AuthorizerIntegrationTest` failures overnight and I found an issue with the `latestVersionUnstable` flag. Let me try to explain.
The `latestVersionUnstable` is used on the broker side to ensure that an unreleased/unstable version is not exposed by the broker. That's fine. However, it does not guarantee that a client having an unreleased/unstable version is not going to use it.
Let's take this change as an example. The version 9 will be shipped in the next release even if we don't want to use it because the schema is there. So the client knows about it and may use it if the broker eventually supports the version. The issue is that the release version my be different from the one shipped so the client would get an error.
I have updated the `AbstractRequest.Builder` to be defensive and only consider stable versions. If the user wants to construct a request for an unstable version, it has to specify it explicitly. This guarantee that even if an unstable version is shipped, it is never used.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14046:
URL: https://github.com/apache/kafka/pull/14046#issuecomment-1642425069
> > Note the use of "latestVersionUnstable": true in the request schema. This means that this new version is not available yet unless activated.
>
> This also leaves us room to add the topic ID to this version if we so choose?
Yeah, that's right.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] jolshan commented on a diff in pull request #14046: KAFKA-14499: [1/N] Introduce OffsetCommit API version 9 and add new StaleMemberEpochException error
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #14046:
URL: https://github.com/apache/kafka/pull/14046#discussion_r1268350437
##########
clients/src/main/java/org/apache/kafka/common/requests/OffsetCommitRequest.java:
##########
@@ -89,33 +88,29 @@ public Map<TopicPartition, Long> offsets() {
return offsets;
}
- public static List<OffsetCommitResponseTopic> getErrorResponseTopics(
- List<OffsetCommitRequestTopic> requestTopics,
Review Comment:
Did we always just pass in all the topics here? I see the usage below that was removed, but curious if we would ever want to make a response for just some topics.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org