You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "AndrewJSchofield (via GitHub)" <gi...@apache.org> on 2026/05/02 21:53:45 UTC
[PR] KAFKA-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
AndrewJSchofield opened a new pull request, #22197:
URL: https://github.com/apache/kafka/pull/22197
This is the implementation of https://cwiki.apache.org/confluence/display/KAFKA/KIP-1323%3A+Initialization+of+share+group+offsets+from+a+specific+offset+or+a+file.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #22197:
URL: https://github.com/apache/kafka/pull/22197#issuecomment-4706101183
> Thanks for the PR and feature @AndrewJSchofield. I tested it and have some questions. I ll test once again with more scenarios once you help with below questions
>
> 1. Should we have share groups output for reset-offsets on consumer-group cli as well? Also, I see a difference when the offsets are fetched from consumer-groups.sh vs share-groups.sh
>
> ```
> kafka git:(KAFKA-20524) bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --dry-run --topic T1
> T1,0,10
>
> ➜ kafka git:(KAFKA-20524) bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --dry-run --topic T1
> T1,0,9
> ```
>
> When there was only a share group and no consumer group.
>
> ```
> kafka git:(KAFKA-20524) ./bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --list
> grp_1_share
> ➜ kafka git:(KAFKA-20524) ./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list
> ```
>
> 2. I reset offset to 2 and started console share consumer again which consumed till offset 10 but start offset didn't move while describing and neither when I exported the offsets. I stopped consumer and restarted but it didn't re-consumed messages from offset 2 rather keep waiting for new data, which was correct. But the outputs were incorrect.
>
> ```
> kafka git:(KAFKA-20524) bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --topic T1:0 --to-offset 2 --execute
>
> GROUP TOPIC PARTITION NEW-OFFSET
> grp_1_share T1 0 2
> ```
>
> Consumed from offset 2:
>
> ```
> kafka git:(KAFKA-20524) ./bin/kafka-console-share-consumer.sh --bootstrap-server localhost:9092 --topic T1 --group grp_1_share
> 3
> 4
> 5
> 6
> 7
> 8
> 9
> 10
> ^CProcessed a total of 8 messages
> ```
>
> But listed start offset incorrectly:
>
> ```
> kafka git:(KAFKA-20524) ./bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --group grp_1_share --describe
>
> GROUP TOPIC PARTITION START-OFFSET LAG
> grp_1_share T1 0 2 0
> ```
>
> Output of export was incorrect as well:
>
> ```
> kafka git:(KAFKA-20524) bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --dry-run --topic T1
> T1,0,2
> ```
>
> Restarted and it didn't consume any old message from offset 2.
>
> ```
> ➜ kafka git:(KAFKA-20524) ./bin/kafka-console-share-consumer.sh --bootstrap-server localhost:9092 --topic T1 --group grp_1_share
> ```
>
> Once produced another message and checked the output as new message was consumed by console consumer then the start offset and export displayed correct data.
>
> 3. Resetting offsets from unknown file provides complete stack trace with correct error. However do we need the stack trace or just the right error as string?
>
> ```
> kafka git:(KAFKA-20524) bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --from-file a --execute
>
> Error: Executing share group command failed due to java.io.IOException: Unable to read file a
> java.lang.RuntimeException: java.io.IOException: Unable to read file a
> at org.apache.kafka.tools.GroupOffsetsResetter.resetPlanFromFile(GroupOffsetsResetter.java:105)
> at org.apache.kafka.tools.GroupOffsetsResetter.resetFromFile(GroupOffsetsResetter.java:389)
> at org.apache.kafka.tools.consumer.group.ShareGroupCommand$ShareGroupService.prepareOffsetsToReset(ShareGroupCommand.java:535)
> at org.apache.kafka.tools.consumer.group.ShareGroupCommand$ShareGroupService.resetOffsetsForInactiveGroup(ShareGroupCommand.java:465)
> at org.apache.kafka.tools.consumer.group.ShareGroupCommand$ShareGroupService.resetOffsets(ShareGroupCommand.java:439)
> at org.apache.kafka.tools.consumer.group.ShareGroupCommand.run(ShareGroupCommand.java:104)
> at org.apache.kafka.tools.consumer.group.ShareGroupCommand.main(ShareGroupCommand.java:89)
> Caused by: java.io.IOException: Unable to read file a
> at org.apache.kafka.common.utils.Utils.readFileAsString(Utils.java:771)
> at org.apache.kafka.tools.GroupOffsetsResetter.resetPlanFromFile(GroupOffsetsResetter.java:101)
> ... 6 more
> Caused by: java.nio.file.NoSuchFileException: a
> at java.base/sun.nio.fs.UnixException.translateToIOException(UnixException.java:92)
> at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:106)
> at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:111)
> at java.base/sun.nio.fs.UnixFileSystemProvider.newByteChannel(UnixFileSystemProvider.java:262)
> at java.base/java.nio.file.Files.newByteChannel(Files.java:380)
> at java.base/java.nio.file.Files.newByteChannel(Files.java:432)
> at java.base/java.nio.file.Files.readAllBytes(Files.java:3292)
> at org.apache.kafka.common.utils.Utils.readFileAsString(Utils.java:768)
> ... 7 more
> ```
>
> 4. What's the difference between `--dry-run` vs `--execute` when we are exporting the offsets with `--export`. Both seems to do the same thing, do we need both? I somewhere remember we discussed it before but was not sure hence thought to write.
>
> ```
> ➜ kafka git:(KAFKA-20524) bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --dry-run --topic T1
> T1,0,10
>
> ➜ kafka git:(KAFKA-20524) bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --execute --topic T1
> T1,0,10
> ```
>
> 5. While running the same commands from 4 using kafka-consumer-groups.sh but with share group then for --execute there is the correct error but with --dry-run it succeeds:
>
> ```
> ➜ kafka git:(KAFKA-20524) bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --execute --topic T1
>
> Error: Executing consumer group command failed due to Failed altering group offsets for the following partitions: [T1-0]
> org.apache.kafka.common.errors.GroupIdNotFoundException: Failed altering group offsets for the following partitions: [T1-0]
>
> ➜ kafka git:(KAFKA-20524) bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --dry-run --topic T1
> T1,0,11
> ```
>
> 6. After resetting the offset there is a difference in output from different clis:
>
> ```
> ➜ kafka git:(KAFKA-20524) ✗ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --from-file a --execute
>
> GROUP TOPIC PARTITION NEW-OFFSET
> grp_1_share T1 0 2
> ➜ kafka git:(KAFKA-20524) ✗ bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --dry-run --topic T1
> T1,0,11
>
> ➜ kafka git:(KAFKA-20524) ✗ bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group grp_1_share --to-current --export --dry-run --topic T1
> T1,0,2
> ```
Thanks for the review @apoorvmittal10.
1. When you use `--dry-run`, the tools do not actually make changes to the group. It is only when you attempt the `--execute` on a consumer group when that group ID is already taken by a share group that the `bin/kafka-consumer-groups.sh` tool complains.
2. This is because share groups lazily update the start offset. That's all you are seeing here.
3. This code is shared with the `bin/kafka-consumer-groups.sh` tool. If you'd like me to change the exception handling for them both, let me know. Personally, I don't mind that much because it is easy enough to tell what's wrong and I prefer not to update existing code unnecessarily because of the risk of introducing unforeseen problems.
4. The `--export` option just tells the tools (both of them) how to output the offset information. If you do `--reset-offsets --dry-run --export`, the group will not be reset, the offsets information for the reset operation is displayed, and the output is made as a comma-separated list which can be redirected to a file. If you do `--reset-offsets --execute --export`, the group will be reset, the offsets information for the reset operation is displayed, and the output is made as a comma-separated list which can be redirected to a file.
What I'd like to do here is open a new KIP which changes how the export works for all group types. It would:
* Introduce `--export-offsets` as a new top-level option. This just obtains the offsets and outputs as a comma-separated list.
* Deprecate `--export` from `--reset-offsets` in AK 4.5, to be removed in AK 5.0.
5. That's because `--dry-run` checks whether there is a non-empty group of the same type as the command being used. This means you can export offsets using the consumer group tool when there is an active share group, and you can export offsets using the share group tool when there is an active consumer group. The first of these is a pre-existing bug.
6. I think this related to the lack of a group type check. The consumer group command looks for a consumer group with the provided group ID, doesn't find one and obtains the information for the topic.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #22197:
URL: https://github.com/apache/kafka/pull/22197#discussion_r3177949649
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -448,14 +483,32 @@ private Collection<TopicPartition> getPartitionsToReset(String groupId) throws E
return partitionsToReset;
}
- private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(Collection<TopicPartition> partitionsToReset) {
+ private Map<TopicPartition, SharePartitionOffsetInfo> getOffsetInfo(String groupId) {
+ try {
+ return adminClient.listShareGroupOffsets(
+ Map.of(groupId, new ListShareGroupOffsetsSpec()),
+ withTimeoutMs(new ListShareGroupOffsetsOptions())
+ ).partitionsToOffsetInfo(groupId).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId, Collection<TopicPartition> partitionsToReset) {
offsetsUtils.checkAllTopicPartitionsValid(partitionsToReset);
- if (opts.options.has(opts.resetToEarliestOpt)) {
+ if (opts.options.has(opts.resetToOffsetOpt)) {
+ return offsetsUtils.resetToOffset(partitionsToReset);
Review Comment:
Not related to this pr, but this call sets offset to 0L, if opts.resetToOffsetOpt is empty or null. Is that ok instead of throwing an exception?
This is also the same in CosnumerGroupCommand too I think.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #22197:
URL: https://github.com/apache/kafka/pull/22197#issuecomment-4366753913
> @AndrewJSchofield thanks for the pr. I have some minor comments.
Thanks for the review. I basically mapped the consumer group tool options across to the share group tool, and you're quite right that there are a few things to improve. I'll sort it out.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22197:
URL: https://github.com/apache/kafka/pull/22197#discussion_r3184194828
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -448,14 +483,32 @@ private Collection<TopicPartition> getPartitionsToReset(String groupId) throws E
return partitionsToReset;
}
- private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(Collection<TopicPartition> partitionsToReset) {
+ private Map<TopicPartition, SharePartitionOffsetInfo> getOffsetInfo(String groupId) {
+ try {
+ return adminClient.listShareGroupOffsets(
+ Map.of(groupId, new ListShareGroupOffsetsSpec()),
+ withTimeoutMs(new ListShareGroupOffsetsOptions())
+ ).partitionsToOffsetInfo(groupId).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId, Collection<TopicPartition> partitionsToReset) {
offsetsUtils.checkAllTopicPartitionsValid(partitionsToReset);
- if (opts.options.has(opts.resetToEarliestOpt)) {
+ if (opts.options.has(opts.resetToOffsetOpt)) {
+ return offsetsUtils.resetToOffset(partitionsToReset);
Review Comment:
I think that's a tiny piece of defensive code in `OffsetsUtils.resetToOffset`. In practice, the command line argument spec for `resetToOffsetOpt` requires an argument of type Long. If no argument or a non-Long argument is specified with `--to-offset`, the command does fail.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #22197:
URL: https://github.com/apache/kafka/pull/22197#discussion_r3183978910
##########
tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java:
##########
@@ -439,6 +439,42 @@ public Map<TopicPartition, OffsetAndMetadata> resetToCurrent(Collection<TopicPar
return preparedOffsetsForPartitionsWithCommittedOffset;
}
+ public Map<TopicPartition, OffsetAndMetadata> resetToCurrentForShareGroup(Collection<TopicPartition> partitionsToReset, Map<TopicPartition, SharePartitionOffsetInfo> currentOffsetInfo) {
Review Comment:
This KIP is still in voting and I'm getting the PR ready in anticipation. Entirely happy to iterate on it.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #22197:
URL: https://github.com/apache/kafka/pull/22197
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22197:
URL: https://github.com/apache/kafka/pull/22197#discussion_r3180752625
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandOptions.java:
##########
@@ -48,14 +48,19 @@ public class ShareGroupCommandOptions extends CommandDefaultOptions {
private static final String COMMAND_CONFIG_DOC = "Property file containing configs to be passed to Admin Client.";
private static final String RESET_OFFSETS_DOC = "Reset offsets of share group. Supports one share group at the time, and instances must be inactive." + NL +
"Has 2 execution options: --dry-run to plan which offsets to reset, and --execute to reset the offsets. " + NL +
- "You must choose one of the following reset specifications: --to-datetime, --to-earliest, --to-latest." + NL +
- "To define the scope use --all-topics or --topic." + NL +
+ "Additionally, the --export option is used to export the offsets in CSV format." + NL +
+ "You must choose one of the following reset specifications: --to-datetime, --to-earliest, --to-latest, --from-file, --to-current, --to-offset." + NL +
+ "To define the scope use --all-topics or --topic. One scope must be specified unless you use '--from-file'." + NL +
Review Comment:
> To define the scope use --all-topics or --topic. One scope must be specified unless you use '--from-file'
It looks like we should remove the condition `if (!options.has(topicOpt) && !options.has(allTopicsOpt))` from line 226, right?
##########
tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java:
##########
@@ -439,6 +439,42 @@ public Map<TopicPartition, OffsetAndMetadata> resetToCurrent(Collection<TopicPar
return preparedOffsetsForPartitionsWithCommittedOffset;
}
+ public Map<TopicPartition, OffsetAndMetadata> resetToCurrentForShareGroup(Collection<TopicPartition> partitionsToReset, Map<TopicPartition, SharePartitionOffsetInfo> currentOffsetInfo) {
Review Comment:
I understand this implementation aligns with `resetToCurrent`, but it feels a bit old-school to me. We could simplify it using the following style
```java
public Map<TopicPartition, OffsetAndMetadata> resetToCurrentForShareGroup(Collection<TopicPartition> partitionsToReset, Map<TopicPartition, SharePartitionOffsetInfo> currentOffsetInfo) {
var partitioned = partitionsToReset.stream().collect(Collectors.partitioningBy(currentOffsetInfo::containsKey));
var partitionsToResetWithStartOffset = partitioned.get(true);
var partitionsToResetWithoutStartOffset = partitioned.get(false);
var preparedOffsetsForPartitionsWithStartOffset = partitionsToResetWithStartOffset.stream()
.collect(Collectors.toMap(Function.identity(), topicPartition -> new OffsetAndMetadata(currentOffsetInfo.get(topicPartition).startOffset())));
getLogEndOffsets(partitionsToResetWithoutStartOffset).forEach((tp, logOffsetResult) -> {
if (!(logOffsetResult instanceof OffsetsUtils.LogOffset logOffset)) {
throw new IllegalStateException("Error getting ending offset of topic partition: " + tp);
}
preparedOffsetsForPartitionsWithStartOffset.put(tp, new OffsetAndMetadata(logOffset.value));
});
return preparedOffsetsForPartitionsWithStartOffset;
}
```
##########
tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java:
##########
@@ -439,6 +439,42 @@ public Map<TopicPartition, OffsetAndMetadata> resetToCurrent(Collection<TopicPar
return preparedOffsetsForPartitionsWithCommittedOffset;
}
+ public Map<TopicPartition, OffsetAndMetadata> resetToCurrentForShareGroup(Collection<TopicPartition> partitionsToReset, Map<TopicPartition, SharePartitionOffsetInfo> currentOffsetInfo) {
Review Comment:
BTW, I'm fine with addressing this in a separate PR
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #22197:
URL: https://github.com/apache/kafka/pull/22197#issuecomment-4373921727
> @AndrewJSchofield Out of curiosity, does the KIP's motivation include aligning offset-related options? If so, should we also support the shift-by option?
@chia7712 The motivation was really to enable offsets to be exported from one group and used to initialise another. I'm not really averse to shift-by, but hadn't planned to do it. There is also another difference and that is that you can only reset the offsets on one share group at a time. When resetting offsets for consumer groups, you can list multiple groups or reset all groups at once.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #22197:
URL: https://github.com/apache/kafka/pull/22197#issuecomment-4743247287
@AndrewJSchofield thanks for the great idea. Let's us keep the discussion on the KIP thread.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #22197:
URL: https://github.com/apache/kafka/pull/22197#issuecomment-4727715065
> > That's because --dry-run checks whether there is a non-empty group of the same type as the command being used. This means you can export offsets using the consumer group tool when there is an active share group, and you can export offsets using the share group tool when there is an active consumer group. The first of these is a pre-existing bug
>
> I think this deserves a patch before we dive into the CLI optimization.
KIP-1359 proposes a new Admin method which would enable us to check the existence and type of a group. As it stands, you can describe groups of a known type, but if a group exists but it's a different type, it is reported as GroupIdNotFound. In order to disambiguate a non-existent group and a group ID not found, it is necessary to list the groups, and listing the groups is an expensive operation just to improve an error message. I wonder whether we can get this small part of KIP-1359 (only just created) into AK 4.4 which would sort out all of this class of problems.
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #22197:
URL: https://github.com/apache/kafka/pull/22197#discussion_r3177923867
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -448,14 +483,32 @@ private Collection<TopicPartition> getPartitionsToReset(String groupId) throws E
return partitionsToReset;
}
- private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(Collection<TopicPartition> partitionsToReset) {
+ private Map<TopicPartition, SharePartitionOffsetInfo> getOffsetInfo(String groupId) {
+ try {
+ return adminClient.listShareGroupOffsets(
+ Map.of(groupId, new ListShareGroupOffsetsSpec()),
+ withTimeoutMs(new ListShareGroupOffsetsOptions())
+ ).partitionsToOffsetInfo(groupId).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId, Collection<TopicPartition> partitionsToReset) {
offsetsUtils.checkAllTopicPartitionsValid(partitionsToReset);
- if (opts.options.has(opts.resetToEarliestOpt)) {
+ if (opts.options.has(opts.resetToOffsetOpt)) {
+ return offsetsUtils.resetToOffset(partitionsToReset);
+ } else if (opts.options.has(opts.resetToEarliestOpt)) {
return offsetsUtils.resetToEarliest(partitionsToReset);
} else if (opts.options.has(opts.resetToLatestOpt)) {
return offsetsUtils.resetToLatest(partitionsToReset);
} else if (opts.options.has(opts.resetToDatetimeOpt)) {
return offsetsUtils.resetToDateTime(partitionsToReset);
+ } else if (offsetsUtils.resetPlanFromFile().isPresent()) {
Review Comment:
This call resetPlanFromFile and next call resetFromFile seem to do same reading and parsing the entire csv. May be use opts to check for presence?
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "muralibasani (via GitHub)" <gi...@apache.org>.
muralibasani commented on code in PR #22197:
URL: https://github.com/apache/kafka/pull/22197#discussion_r3177937861
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -448,14 +483,32 @@ private Collection<TopicPartition> getPartitionsToReset(String groupId) throws E
return partitionsToReset;
}
- private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(Collection<TopicPartition> partitionsToReset) {
+ private Map<TopicPartition, SharePartitionOffsetInfo> getOffsetInfo(String groupId) {
+ try {
+ return adminClient.listShareGroupOffsets(
+ Map.of(groupId, new ListShareGroupOffsetsSpec()),
+ withTimeoutMs(new ListShareGroupOffsetsOptions())
+ ).partitionsToOffsetInfo(groupId).get();
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
Review Comment:
Is it better to interrupt the currentThread for RunTimeException?
For ExecutionException, getting the instance type for ex if it's of KafkaException(GroupAuthorizationException etc), it could be more visible to the caller ?
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #22197:
URL: https://github.com/apache/kafka/pull/22197#issuecomment-4720802348
> What I'd like to do here is open a new KIP which changes how the export works for all group types. It would:
Have you created the Jira ticket? I think `--export` could be renamed to `--format` as well
--
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-20524: CSV reset offset plan for kafka-share-groups.sh [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #22197:
URL: https://github.com/apache/kafka/pull/22197#issuecomment-4720927686
> That's because --dry-run checks whether there is a non-empty group of the same type as the command being used. This means you can export offsets using the consumer group tool when there is an active share group, and you can export offsets using the share group tool when there is an active consumer group. The first of these is a pre-existing bug
I think this deserves a patch before we dive into the CLI optimization.
--
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