You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "sstremler (via GitHub)" <gi...@apache.org> on 2026/04/17 23:07:38 UTC
[PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
sstremler opened a new pull request, #22087:
URL: https://github.com/apache/kafka/pull/22087
Prevents `kafka-consumer-groups.sh --reset-offsets` from modifying streams groups by detecting them early and returning an error. Tests are added to verify the new behavior.
Reviewers: Matthias J. Sax [matthias@confluent.io](mailto:matthias@confluent.io)
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "sstremler (via GitHub)" <gi...@apache.org>.
sstremler commented on code in PR #22087:
URL: https://github.com/apache/kafka/pull/22087#discussion_r3109655559
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -269,8 +269,16 @@ private void printGroupInfo(List<GroupListing> groups, boolean includeType, bool
}
List<String> listConsumerGroups() {
+ return listGroups(ListGroupsOptions.forConsumerGroups());
+ }
+
+ List<String> listStreamGroups() {
+ return listGroups(ListGroupsOptions.forStreamsGroups());
+ }
+
+ List<String> listGroups(ListGroupsOptions options) {
try {
- ListGroupsResult result = adminClient.listGroups(withTimeoutMs(ListGroupsOptions.forConsumerGroups()));
+ ListGroupsResult result = adminClient.listGroups(withTimeoutMs(options));
Collection<GroupListing> listings = result.all().get();
Review Comment:
`describeConsumerGroups(...).get()` throws `GroupIdNotFoundException` when it's called for a streams group. So in this case calling `ConsumerGroupDescription.type()` as you suggested is not possible.
The case when `GroupIdNotFoundException` is thrown is handled the same as the case when group id is found and inactive. Do you know why?
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #22087:
URL: https://github.com/apache/kafka/pull/22087#discussion_r3110871441
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -269,8 +269,16 @@ private void printGroupInfo(List<GroupListing> groups, boolean includeType, bool
}
List<String> listConsumerGroups() {
+ return listGroups(ListGroupsOptions.forConsumerGroups());
+ }
+
+ List<String> listStreamGroups() {
+ return listGroups(ListGroupsOptions.forStreamsGroups());
+ }
+
+ List<String> listGroups(ListGroupsOptions options) {
try {
- ListGroupsResult result = adminClient.listGroups(withTimeoutMs(ListGroupsOptions.forConsumerGroups()));
+ ListGroupsResult result = adminClient.listGroups(withTimeoutMs(options));
Collection<GroupListing> listings = result.all().get();
Review Comment:
ohh,in that case the earlier suggestion would not work.describeConsumerGroups only queries consumer groups. Streams ids are not consumer groups for that API, so the coordinator returns GROUP_ID_NOT_FOUND → GroupIdNotFoundException on get();
As per the current change, we are returning before groupDescription.get().groupState().toString(); this so we should be fine.
but we may still get into issue in listGroups for old broker. we will have to handle that case.
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22087:
URL: https://github.com/apache/kafka/pull/22087#issuecomment-5018866920
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #22087:
URL: https://github.com/apache/kafka/pull/22087#issuecomment-6307496324
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #22087: KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh
URL: https://github.com/apache/kafka/pull/22087
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "nileshkumar3 (via GitHub)" <gi...@apache.org>.
nileshkumar3 commented on code in PR #22087:
URL: https://github.com/apache/kafka/pull/22087#discussion_r3108089112
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -269,8 +269,16 @@ private void printGroupInfo(List<GroupListing> groups, boolean includeType, bool
}
List<String> listConsumerGroups() {
+ return listGroups(ListGroupsOptions.forConsumerGroups());
+ }
+
+ List<String> listStreamGroups() {
+ return listGroups(ListGroupsOptions.forStreamsGroups());
+ }
+
+ List<String> listGroups(ListGroupsOptions options) {
try {
- ListGroupsResult result = adminClient.listGroups(withTimeoutMs(ListGroupsOptions.forConsumerGroups()));
+ ListGroupsResult result = adminClient.listGroups(withTimeoutMs(options));
Collection<GroupListing> listings = result.all().get();
Review Comment:
We already call `describeConsumerGroups(...)` in `resetOffsets()`, and `ConsumerGroupDescription.type()` can identify a streams group, so we can skip an extra `listGroups(forStreamsGroups())`. That saves a round trip and avoids relying on `LIST_GROUPS` with a STREAMS type filter, which requires newer broker support and may fail on older brokers.
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "sstremler (via GitHub)" <gi...@apache.org>.
sstremler commented on code in PR #22087:
URL: https://github.com/apache/kafka/pull/22087#discussion_r3112517669
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -269,8 +269,16 @@ private void printGroupInfo(List<GroupListing> groups, boolean includeType, bool
}
List<String> listConsumerGroups() {
+ return listGroups(ListGroupsOptions.forConsumerGroups());
+ }
+
+ List<String> listStreamGroups() {
+ return listGroups(ListGroupsOptions.forStreamsGroups());
+ }
+
+ List<String> listGroups(ListGroupsOptions options) {
try {
- ListGroupsResult result = adminClient.listGroups(withTimeoutMs(ListGroupsOptions.forConsumerGroups()));
+ ListGroupsResult result = adminClient.listGroups(withTimeoutMs(options));
Collection<GroupListing> listings = result.all().get();
Review Comment:
I'm waiting for someone else to confirm that the 4.X `kafka-consumer-groups.sh` needs to work with 3.X brokers as you suggested.
--
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
Re: [PR] KAFKA-20454: Prevent offset reset for streams groups in kafka-consumer-groups.sh [kafka]
Posted by "sstremler (via GitHub)" <gi...@apache.org>.
sstremler commented on code in PR #22087:
URL: https://github.com/apache/kafka/pull/22087#discussion_r3109655559
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ConsumerGroupCommand.java:
##########
@@ -269,8 +269,16 @@ private void printGroupInfo(List<GroupListing> groups, boolean includeType, bool
}
List<String> listConsumerGroups() {
+ return listGroups(ListGroupsOptions.forConsumerGroups());
+ }
+
+ List<String> listStreamGroups() {
+ return listGroups(ListGroupsOptions.forStreamsGroups());
+ }
+
+ List<String> listGroups(ListGroupsOptions options) {
try {
- ListGroupsResult result = adminClient.listGroups(withTimeoutMs(ListGroupsOptions.forConsumerGroups()));
+ ListGroupsResult result = adminClient.listGroups(withTimeoutMs(options));
Collection<GroupListing> listings = result.all().get();
Review Comment:
`describeConsumerGroups(...).get()` throws `GroupIdNotFoundException` when it's called for a streams group. The case when `GroupIdNotFoundException` is thrown is handled the same as the case when group id is found and inactive. Do you know why?
--
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