You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "JimmyWang6 (via GitHub)" <gi...@apache.org> on 2025/05/27 01:39:51 UTC
[PR] KAFKA-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
JimmyWang6 opened a new pull request, #19820:
URL: https://github.com/apache/kafka/pull/19820
[KAFKA-16720](https://issues.apache.org/jira/browse/KAFKA-16720) aims to
finish the AlterShareGroupOffsets for ShareGroupCommand part.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3049468549
@AndrewJSchofield So much thanks for your help to resolving the issues. I sincerely apologize for the previous bugs I introduced. I’ll resolve the conflicts and thoroughly test the script as soon as possible to ensure everything works smoothly.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2234136591
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
@JimmyWang6 I see that the issue in #20168 is still occurring. I'm not sure that you need PR #20235 to merge before you can address the problem of a long delay if resetting offsets for a non-existent partition.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2215897015
##########
tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java:
##########
@@ -69,6 +69,21 @@ public OffsetsUtils(Admin adminClient, OptionParser parser, OffsetsUtilsOptions
this.parser = parser;
}
+ public static void printOffsetsToReset(Map<String, Map<TopicPartition, OffsetAndMetadata>> groupAssignmentsToReset) {
+ String format = "%n%-30s %-30s %-10s %-15s";
Review Comment:
If you look at `ShareGroupCommand.printOffsetsFormat`, you'll see that the column alignment is much neater. I suggest:
* group is `max(15, groupId.length())` characters wide
* topic is `max(15, maxTopicLen)` characters wide
* partition is 10 characters wide
* offset does not specify
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java:
##########
@@ -1052,6 +1059,217 @@ public void testDeleteShareGroupsPartialFailure() {
assertEquals(expectedResults, service.deleteShareGroups());
}
}
+
+ @Test
+ public void testAlterShareGroupAllTopicsSuccess() {
+ String group = "share-group";
+ String topic1 = "topic1";
+ String topic2 = "topic2";
+ String bootstrapServer = "localhost:9092";
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServer, "--reset-offsets", "--to-earliest", "--execute", "--topic", topic1, "--topic", topic2, "--group", group};
+ Admin adminClient = mock(KafkaAdminClient.class);
+
+ ListShareGroupOffsetsResult listShareGroupOffsetsResult = AdminClientTestUtils.createListShareGroupOffsetsResult(
+ Map.of(
+ group,
+ KafkaFuture.completedFuture(Map.of(new TopicPartition(topic1, 0), new OffsetAndMetadata(10L), new TopicPartition(topic1, 1), new OffsetAndMetadata(10L),
+ new TopicPartition(topic2, 0), new OffsetAndMetadata(0L)))
+ )
+ );
+ when(adminClient.listShareGroupOffsets(any())).thenReturn(listShareGroupOffsetsResult);
+ ListTopicsResult listTopicsResult = mock(ListTopicsResult.class);
+ Set<String> topics = Set.of("topic1", "topic2");
+ when(listTopicsResult.names()).thenReturn(completedFuture(topics));
+ when(adminClient.listTopics()).thenReturn(listTopicsResult);
+
+ AlterShareGroupOffsetsResult alterShareGroupOffsetsResult = mockAlterShareGroupOffsets(adminClient, group);
+ Map<TopicPartition, OffsetAndMetadata> partitionOffsets = Map.of(new TopicPartition(topic1, 0), new OffsetAndMetadata(0L), new TopicPartition(topic1, 1), new OffsetAndMetadata(0L),
+ new TopicPartition(topic2, 0), new OffsetAndMetadata(0L));
+ ListOffsetsResult listOffsetsResult = AdminClientTestUtils.createListOffsetsResult(partitionOffsets);
+ when(adminClient.listOffsets(any(), any(ListOffsetsOptions.class))).thenReturn(listOffsetsResult);
+
+ ShareGroupDescription exp = new ShareGroupDescription(
+ group,
+ List.of(),
+ GroupState.EMPTY,
+ new Node(0, "host1", 9090), 0, 0);
+ DescribeShareGroupsResult describeShareGroupsResult = mock(DescribeShareGroupsResult.class);
+ when(describeShareGroupsResult.describedGroups()).thenReturn(Map.of(group, KafkaFuture.completedFuture(exp)));
+ when(adminClient.describeShareGroups(ArgumentMatchers.anyCollection(), any(DescribeShareGroupsOptions.class))).thenReturn(describeShareGroupsResult);
+ try (ShareGroupService service = getShareGroupService(cgcArgs, adminClient)) {
+ service.resetOffsets();
+ verify(adminClient).alterShareGroupOffsets(eq(group), anyMap());
+ verify(alterShareGroupOffsetsResult, times(1)).all();
+ verify(adminClient).describeShareGroups(ArgumentMatchers.anyCollection(), any(DescribeShareGroupsOptions.class));
+ }
+ }
+
+ @Test
+ public void testResetOffsetsDryRunSuccess() {
+ String group = "share-group";
+ String topic = "topic";
+ String bootstrapServer = "localhost:9092";
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServer, "--reset-offsets", "--to-earliest", "--dry-run", "--topic", topic, "--group", group};
Review Comment:
Let's have some tests for `--to-latest` and `--to-datetime`.
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java:
##########
@@ -1052,6 +1059,217 @@ public void testDeleteShareGroupsPartialFailure() {
assertEquals(expectedResults, service.deleteShareGroups());
}
}
+
+ @Test
+ public void testAlterShareGroupAllTopicsSuccess() {
Review Comment:
This is testing multiple topics, not `--all-topics`. Both are valid.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-2994395160
@JimmyWang6 could you please fix the build error?
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2160238278
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +428,70 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ )),
+ new AlterShareGroupOffsetsOptions()
+ .timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = new HashMap<>();
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = new HashSet<>(opts.options.valuesOf(opts.topicOpt));
Review Comment:
Good idea.Updated.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2163756146
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +428,70 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ )),
+ new AlterShareGroupOffsetsOptions()
+ .timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
Review Comment:
Good catch. I’ve fixed the bug by adding validation to ensure that either --all-topics or --topic is provided.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2215935427
##########
tools/src/main/java/org/apache/kafka/tools/OffsetsUtils.java:
##########
@@ -69,6 +69,21 @@ public OffsetsUtils(Admin adminClient, OptionParser parser, OffsetsUtilsOptions
this.parser = parser;
}
+ public static void printOffsetsToReset(Map<String, Map<TopicPartition, OffsetAndMetadata>> groupAssignmentsToReset) {
+ String format = "%n%-30s %-30s %-10s %-15s";
Review Comment:
Copy 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
Re: [PR] KAFKA-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2207999338
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,67 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = Set.copyOf(opts.options.valuesOf(opts.topicOpt));
+ Set<String> existsTopics = offsetsByTopicPartitions.keySet().stream()
+ .map(TopicPartition::topic)
+ .collect(Collectors.toSet());
+ if (!existsTopics.containsAll(topics)) {
+ CommandLineUtils
+ .printUsageAndExit(opts.parser, String.format("Share group '%s' is not subscribed to topic '%s'",
Review Comment:
I agree with your viewpoint. I've already added validation for non-existent topics, which require an extra `adminClient.listTopics()` request.
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,67 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = Set.copyOf(opts.options.valuesOf(opts.topicOpt));
+ Set<String> existsTopics = offsetsByTopicPartitions.keySet().stream()
+ .map(TopicPartition::topic)
+ .collect(Collectors.toSet());
+ if (!existsTopics.containsAll(topics)) {
+ CommandLineUtils
+ .printUsageAndExit(opts.parser, String.format("Share group '%s' is not subscribed to topic '%s'",
Review Comment:
I agree with your point. I've already added validation for non-existent topics, which require an extra `adminClient.listTopics()` request.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3089311546
Please also merge latest changes from trunk. I expect the build will be clean afterwards.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3089184428
Still getting the usage message for trying to reset offsets for a consumer group.
```
bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group CG1 --topic T1 --execute --to-earliest
Group CG1 is not a share group.
Option Description
------ -----------
--all-groups Apply to all share groups.
--all-topics Consider all topics assigned to a
share group in the 'reset-offsets'
process
```
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2221352615
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
Review Comment:
Sorry, I accidentally missed this issue earlier, but it has now been fixed. Thank you for bringing it to my attention!
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3031674467
@JimmyWang6 Please merge latest changes. Now that https://github.com/apache/kafka/pull/20049 is merged, the `Admin.alterShareGroupOffsets` will now return top-level errors correctly. This means `kafka-share-groups.sh --reset-offsets` in error cases will fail, rather than appearing to work when it did not.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2160238688
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +428,70 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ )),
+ new AlterShareGroupOffsetsOptions()
+ .timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = new HashMap<>();
Review Comment:
Good catch, thanks.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2220754299
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
Review Comment:
I thinks we should, fixed.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2207999338
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,67 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = Set.copyOf(opts.options.valuesOf(opts.topicOpt));
+ Set<String> existsTopics = offsetsByTopicPartitions.keySet().stream()
+ .map(TopicPartition::topic)
+ .collect(Collectors.toSet());
+ if (!existsTopics.containsAll(topics)) {
+ CommandLineUtils
+ .printUsageAndExit(opts.parser, String.format("Share group '%s' is not subscribed to topic '%s'",
Review Comment:
I agree with your viewpoint. I've already added validation for non-existent 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
Re: [PR] KAFKA-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2217373514
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java:
##########
@@ -1052,6 +1059,217 @@ public void testDeleteShareGroupsPartialFailure() {
assertEquals(expectedResults, service.deleteShareGroups());
}
}
+
+ @Test
+ public void testAlterShareGroupAllTopicsSuccess() {
+ String group = "share-group";
+ String topic1 = "topic1";
+ String topic2 = "topic2";
+ String bootstrapServer = "localhost:9092";
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServer, "--reset-offsets", "--to-earliest", "--execute", "--topic", topic1, "--topic", topic2, "--group", group};
+ Admin adminClient = mock(KafkaAdminClient.class);
+
+ ListShareGroupOffsetsResult listShareGroupOffsetsResult = AdminClientTestUtils.createListShareGroupOffsetsResult(
+ Map.of(
+ group,
+ KafkaFuture.completedFuture(Map.of(new TopicPartition(topic1, 0), new OffsetAndMetadata(10L), new TopicPartition(topic1, 1), new OffsetAndMetadata(10L),
+ new TopicPartition(topic2, 0), new OffsetAndMetadata(0L)))
+ )
+ );
+ when(adminClient.listShareGroupOffsets(any())).thenReturn(listShareGroupOffsetsResult);
+ ListTopicsResult listTopicsResult = mock(ListTopicsResult.class);
+ Set<String> topics = Set.of("topic1", "topic2");
+ when(listTopicsResult.names()).thenReturn(completedFuture(topics));
+ when(adminClient.listTopics()).thenReturn(listTopicsResult);
+
+ AlterShareGroupOffsetsResult alterShareGroupOffsetsResult = mockAlterShareGroupOffsets(adminClient, group);
+ Map<TopicPartition, OffsetAndMetadata> partitionOffsets = Map.of(new TopicPartition(topic1, 0), new OffsetAndMetadata(0L), new TopicPartition(topic1, 1), new OffsetAndMetadata(0L),
+ new TopicPartition(topic2, 0), new OffsetAndMetadata(0L));
+ ListOffsetsResult listOffsetsResult = AdminClientTestUtils.createListOffsetsResult(partitionOffsets);
+ when(adminClient.listOffsets(any(), any(ListOffsetsOptions.class))).thenReturn(listOffsetsResult);
+
+ ShareGroupDescription exp = new ShareGroupDescription(
+ group,
+ List.of(),
+ GroupState.EMPTY,
+ new Node(0, "host1", 9090), 0, 0);
+ DescribeShareGroupsResult describeShareGroupsResult = mock(DescribeShareGroupsResult.class);
+ when(describeShareGroupsResult.describedGroups()).thenReturn(Map.of(group, KafkaFuture.completedFuture(exp)));
+ when(adminClient.describeShareGroups(ArgumentMatchers.anyCollection(), any(DescribeShareGroupsOptions.class))).thenReturn(describeShareGroupsResult);
+ try (ShareGroupService service = getShareGroupService(cgcArgs, adminClient)) {
+ service.resetOffsets();
+ verify(adminClient).alterShareGroupOffsets(eq(group), anyMap());
+ verify(alterShareGroupOffsetsResult, times(1)).all();
+ verify(adminClient).describeShareGroups(ArgumentMatchers.anyCollection(), any(DescribeShareGroupsOptions.class));
+ }
+ }
+
+ @Test
+ public void testResetOffsetsDryRunSuccess() {
+ String group = "share-group";
+ String topic = "topic";
+ String bootstrapServer = "localhost:9092";
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServer, "--reset-offsets", "--to-earliest", "--dry-run", "--topic", topic, "--group", group};
Review Comment:
I have added `testAlterShareGroupToLatestSuccess()` and `testAlterShareGroupAllTopicsToDatetimeSuccess()` tests for those two cases, PTAL :)
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2217373514
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java:
##########
@@ -1052,6 +1059,217 @@ public void testDeleteShareGroupsPartialFailure() {
assertEquals(expectedResults, service.deleteShareGroups());
}
}
+
+ @Test
+ public void testAlterShareGroupAllTopicsSuccess() {
+ String group = "share-group";
+ String topic1 = "topic1";
+ String topic2 = "topic2";
+ String bootstrapServer = "localhost:9092";
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServer, "--reset-offsets", "--to-earliest", "--execute", "--topic", topic1, "--topic", topic2, "--group", group};
+ Admin adminClient = mock(KafkaAdminClient.class);
+
+ ListShareGroupOffsetsResult listShareGroupOffsetsResult = AdminClientTestUtils.createListShareGroupOffsetsResult(
+ Map.of(
+ group,
+ KafkaFuture.completedFuture(Map.of(new TopicPartition(topic1, 0), new OffsetAndMetadata(10L), new TopicPartition(topic1, 1), new OffsetAndMetadata(10L),
+ new TopicPartition(topic2, 0), new OffsetAndMetadata(0L)))
+ )
+ );
+ when(adminClient.listShareGroupOffsets(any())).thenReturn(listShareGroupOffsetsResult);
+ ListTopicsResult listTopicsResult = mock(ListTopicsResult.class);
+ Set<String> topics = Set.of("topic1", "topic2");
+ when(listTopicsResult.names()).thenReturn(completedFuture(topics));
+ when(adminClient.listTopics()).thenReturn(listTopicsResult);
+
+ AlterShareGroupOffsetsResult alterShareGroupOffsetsResult = mockAlterShareGroupOffsets(adminClient, group);
+ Map<TopicPartition, OffsetAndMetadata> partitionOffsets = Map.of(new TopicPartition(topic1, 0), new OffsetAndMetadata(0L), new TopicPartition(topic1, 1), new OffsetAndMetadata(0L),
+ new TopicPartition(topic2, 0), new OffsetAndMetadata(0L));
+ ListOffsetsResult listOffsetsResult = AdminClientTestUtils.createListOffsetsResult(partitionOffsets);
+ when(adminClient.listOffsets(any(), any(ListOffsetsOptions.class))).thenReturn(listOffsetsResult);
+
+ ShareGroupDescription exp = new ShareGroupDescription(
+ group,
+ List.of(),
+ GroupState.EMPTY,
+ new Node(0, "host1", 9090), 0, 0);
+ DescribeShareGroupsResult describeShareGroupsResult = mock(DescribeShareGroupsResult.class);
+ when(describeShareGroupsResult.describedGroups()).thenReturn(Map.of(group, KafkaFuture.completedFuture(exp)));
+ when(adminClient.describeShareGroups(ArgumentMatchers.anyCollection(), any(DescribeShareGroupsOptions.class))).thenReturn(describeShareGroupsResult);
+ try (ShareGroupService service = getShareGroupService(cgcArgs, adminClient)) {
+ service.resetOffsets();
+ verify(adminClient).alterShareGroupOffsets(eq(group), anyMap());
+ verify(alterShareGroupOffsetsResult, times(1)).all();
+ verify(adminClient).describeShareGroups(ArgumentMatchers.anyCollection(), any(DescribeShareGroupsOptions.class));
+ }
+ }
+
+ @Test
+ public void testResetOffsetsDryRunSuccess() {
+ String group = "share-group";
+ String topic = "topic";
+ String bootstrapServer = "localhost:9092";
+ String[] cgcArgs = new String[]{"--bootstrap-server", bootstrapServer, "--reset-offsets", "--to-earliest", "--dry-run", "--topic", topic, "--group", group};
Review Comment:
I have added `testAlterShareGroupToLatestSuccess()` and `testAlterShareGroupAllTopicsToDatetimeSuccess()` tests for those two cases.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2165262658
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3772,7 +3772,19 @@ class KafkaApis(val requestChannel: RequestChannel,
case Some(error) =>
topic.partitions().forEach(partition => responseBuilder.addPartition(topic.topicName(), partition.partitionIndex(), metadataCache.topicNamesToIds(), error.error))
case None =>
- authorizedTopicPartitions.add(topic)
Review Comment:
Good idea! Updated.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2222326661
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,74 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!(GroupState.EMPTY.equals(shareGroupDescription.groupState()) || GroupState.DEAD.equals(shareGroupDescription.groupState()))) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
Review Comment:
What happened if we can't get response from broker?
Should we add try catch here?
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2221245988
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
Review Comment:
@JimmyWang6 Do you fix this issue?
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2222437696
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,74 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!(GroupState.EMPTY.equals(shareGroupDescription.groupState()) || GroupState.DEAD.equals(shareGroupDescription.groupState()))) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
Review Comment:
Thanks for your comments. I have added a try-catch block at the outer `resetOffsets()` layer. If it's a KafkaException, the errorMessage will be printed directly. For any other exceptions, they will be thrown directly.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2200620277
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,67 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = Set.copyOf(opts.options.valuesOf(opts.topicOpt));
+ Set<String> existsTopics = offsetsByTopicPartitions.keySet().stream()
+ .map(TopicPartition::topic)
+ .collect(Collectors.toSet());
+ if (!existsTopics.containsAll(topics)) {
+ CommandLineUtils
+ .printUsageAndExit(opts.parser, String.format("Share group '%s' is not subscribed to topic '%s'",
Review Comment:
This doesn't distinguish between non-existent topics and topics which are just not subscribed. This is probably a debatable point, so I wonder what you think.
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,67 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
Review Comment:
The `--verbose` option is not valid with `--reset-offsets` so this is a bit weird. But actually, I think it would be better if there was something like the consumer group tool `printOffsetsToReset`. This is just because I think having a column heading `NEW-START-OFFSET` would make it clear that it's the new start offset.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3053638805
Hi @chia7712 @AndrewJSchofield ,
I have updated the PR and addressed the [comments](https://github.com/apache/kafka/pull/18929#pullrequestreview-3002388120), PTAL when get a chance, thanks.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-2987824602
@JimmyWang6 There's a build failure due to spotless check. Please resolve it and merge latest changes.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2218515195
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
@chia7712 Thanks for your comments, I will handle the topic-partions 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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3073471854
Please fix the checkstyle build error.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3092455657
@AndrewJSchofield I have addressed the below comments as your suggestions. Please take a glance to see if they look good. Thanks! :)
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-2938732620
Hi all @AndrewJSchofield @chia7712 @smjn @apoorvmittal10 ，
I have finished the basic part of `kafka-share-groups.sh` --reset-offsets`. PTAL if you have time ;)
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-2974550203
Hi @AndrewJSchofield @smjn @apoorvmittal10 ,
Sorry for the late response, could you please take a look if there is anything wrong with the current approach? Thanks in advance.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3089333514
> Still getting the usage message for trying to reset offsets for a consumer group.
>
> ```
> bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group CG1 --topic T1 --execute --to-earliest
> Group CG1 is not a share group.
> Option Description
> ------ -----------
> --all-groups Apply to all share groups.
> --all-topics Consider all topics assigned to a
> share group in the 'reset-offsets'
> process
> ```
Oops, I'm sorry that I didn't grasp your meaning, I will try to replace `CommandLineUtils.printUsageAndExit()` method call by a simple message.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2220759816
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
Review Comment:
Thanks for pointing it out, this is a little mistake and I have addressed 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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2221245988
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
Review Comment:
@JimmyWang6 Do you fix this issue or I missed something ?
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2156911513
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +428,70 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ )),
+ new AlterShareGroupOffsetsOptions()
+ .timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = new HashMap<>();
Review Comment:
This probably ought to be `groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec())`.
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +428,70 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ )),
+ new AlterShareGroupOffsetsOptions()
+ .timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
Review Comment:
The error handling for `bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --group G1 --reset-offsets --dry-run` needs to be stronger. I didn't specify any topic information so the map `offsetsByTopicPartitions` is empty and the command blows up.
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +428,70 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ )),
+ new AlterShareGroupOffsetsOptions()
+ .timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = new HashMap<>();
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = new HashSet<>(opts.options.valuesOf(opts.topicOpt));
Review Comment:
Maybe `Set.copyOf(opts.options.valuesOf(opts.topicOpt))`
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2220755722
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ Map<Uuid, Map<Integer, PartitionErrorData>> topicPartitionErrorsMap = result.getErrors();
+ topic.partitions().forEach(partition -> {
+ if (partition.errorCode() != Errors.NONE.code()) {
+ topicData.partitions().add(partition);
+ return;
+ }
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = topicPartitionErrorsMap.get(topic.topicId()).get(partition.partitionIndex());
+ if (error == null) {
+ partitionData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition()
Review Comment:
Good idea, I have change to use partition.duplicate() to build the response.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2166261902
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,71 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = Set.copyOf(opts.options.valuesOf(opts.topicOpt));
+ Set<String> existsTopics = offsetsByTopicPartitions.keySet().stream()
+ .map(TopicPartition::topic)
+ .collect(Collectors.toSet());
+ if (!existsTopics.containsAll(topics)) {
+ CommandLineUtils
+ .printUsageAndExit(opts.parser, String.format("Some topics %s do not exist in share group '%s'.",
+ topics.stream().filter(topic -> !existsTopics.contains(topic)).collect(Collectors.joining(", ")), groupId));
+ }
+
+ offsetsByTopicPartitions = offsetsByTopicPartitions.entrySet().stream()
+ .filter(entry -> {
+ TopicPartition topicPartition = entry.getKey();
+ return topics.contains(topicPartition.topic());
+ })
Review Comment:
Updated. Thanks for your review.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -682,15 +682,48 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ Map<Uuid, Map<Integer, PartitionErrorData>> topicPartitionErrorsMap = result.getErrors();
+ topic.partitions().forEach(partition -> {
+ if (partition.errorCode() != Errors.NONE.code()) {
+ topicData.partitions().add(partition);
+ return;
+ }
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = topicPartitionErrorsMap.get(topic.topicId()).get(partition.partitionIndex());
+ if (error == null) {
+ partitionData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition()
+ .setPartitionIndex(partition.partitionIndex())
+ .setErrorCode(Errors.NONE.code());
+ } else {
+ partitionData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition()
+ .setPartitionIndex(partition.partitionIndex())
+ .setErrorCode(error.errorCode())
+ .setErrorMessage(error.errorMessage());
+ }
+ topicData.partitions().add(partitionData);
+ });
+ return topicData;
+ })
+ .iterator()));
+ return data;
+
Review Comment:
Fixed.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2165943288
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,71 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = Set.copyOf(opts.options.valuesOf(opts.topicOpt));
+ Set<String> existsTopics = offsetsByTopicPartitions.keySet().stream()
+ .map(TopicPartition::topic)
+ .collect(Collectors.toSet());
+ if (!existsTopics.containsAll(topics)) {
+ CommandLineUtils
+ .printUsageAndExit(opts.parser, String.format("Some topics %s do not exist in share group '%s'.",
+ topics.stream().filter(topic -> !existsTopics.contains(topic)).collect(Collectors.joining(", ")), groupId));
+ }
+
+ offsetsByTopicPartitions = offsetsByTopicPartitions.entrySet().stream()
+ .filter(entry -> {
+ TopicPartition topicPartition = entry.getKey();
+ return topics.contains(topicPartition.topic());
+ })
Review Comment:
```suggestion
.filter(entry -> topics.contains(entry.getKey().topic()))
```
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -682,15 +682,48 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ Map<Uuid, Map<Integer, PartitionErrorData>> topicPartitionErrorsMap = result.getErrors();
+ topic.partitions().forEach(partition -> {
+ if (partition.errorCode() != Errors.NONE.code()) {
+ topicData.partitions().add(partition);
+ return;
+ }
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = topicPartitionErrorsMap.get(topic.topicId()).get(partition.partitionIndex());
+ if (error == null) {
+ partitionData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition()
+ .setPartitionIndex(partition.partitionIndex())
+ .setErrorCode(Errors.NONE.code());
+ } else {
+ partitionData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition()
+ .setPartitionIndex(partition.partitionIndex())
+ .setErrorCode(error.errorCode())
+ .setErrorMessage(error.errorMessage());
+ }
+ topicData.partitions().add(partitionData);
+ });
+ return topicData;
+ })
+ .iterator()));
+ return data;
+
Review Comment:
nit: extra blank
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-2996332151
Hi @chia7712，
Thanks for your comment and I have updated the PR, PTAL if you have time.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3062032067
Thanks for the updates. I'm starting to review and test it. Here's a small problem I've found so far. `SG1` is an existing share group with 1 member. `T2` doesn't exist, so the message about "Not subscribed" is fine. However, the usage message for the tool is printed even though it's a regular expected error.
```
bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group SG1 --execute --topic T2 --to-earliest
Share group 'SG1' is not subscribed to topic 'T2'
Option Description
------ -----------
--all-groups Apply to all share groups.
(( And so on ))
```
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2164565204
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3772,7 +3772,19 @@ class KafkaApis(val requestChannel: RequestChannel,
case Some(error) =>
topic.partitions().forEach(partition => responseBuilder.addPartition(topic.topicName(), partition.partitionIndex(), metadataCache.topicNamesToIds(), error.error))
case None =>
- authorizedTopicPartitions.add(topic)
Review Comment:
Excuse me, why to make this change? it makes a copy of `topic`
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -682,15 +682,43 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ topic.partitions().forEach(partition -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = result.getErrors().get(topic.topicId()).get(partition.partitionIndex());
Review Comment:
Furthermore, how do we ensure that the topic from `AlterShareGroupOffsetsResponseData` must exist in `InitializeShareGroupStateResult`?
If users attempt to alter the nonexistent partitions within an existing topic, the `AlterShareGroupOffsetsResponseData` will contain the topic but `InitializeShareGroupStateResult` will not. Is this correct?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -682,15 +682,43 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ topic.partitions().forEach(partition -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = result.getErrors().get(topic.topicId()).get(partition.partitionIndex());
Review Comment:
`result.getErrors()` will create a new collection by grouping data. Perhaps it should be moved out of loop.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2164713759
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3772,7 +3772,19 @@ class KafkaApis(val requestChannel: RequestChannel,
case Some(error) =>
topic.partitions().forEach(partition => responseBuilder.addPartition(topic.topicName(), partition.partitionIndex(), metadataCache.topicNamesToIds(), error.error))
case None =>
- authorizedTopicPartitions.add(topic)
Review Comment:
Hi @chia7712 ,
Much thanks for your review.
This is a bug related to my previous PR :( The add method always returns false and fails to add the topic correctly because the condition `newElement.prev() != INVALID_INDEX || newElement.next() != INVALID_INDEX `evaluates to true. I believe this occurs because the topic has already been added to an `ImplicitLinkedHashCollection`, meaning the prev() and next() pointers of the specified topic are no longer invalid. Therefore, I'm attempting to create a new topic here.
public final boolean add(E newElement) {
if (newElement == null) {
return false;
}
if (newElement.prev() != INVALID_INDEX || newElement.next() != INVALID_INDEX) {
return false;
}
if ((size + 1) >= elements.length / 2) {
changeCapacity(calculateCapacity(elements.length));
}
int slot = addInternal(newElement, elements);
if (slot >= 0) {
addToListTail(head, elements, slot);
size++;
return true;
}
return false;
}
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2164757904
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -682,15 +682,43 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ topic.partitions().forEach(partition -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = result.getErrors().get(topic.topicId()).get(partition.partitionIndex());
Review Comment:
Good catch! My idea is if users want to alter one nonexistent partition, the error code and message will be added to `AlterShareGroupOffsetsResponseData` when executing `groupMetadataManager.completeAlterShareGroupOffsets`. So I plan to skip the processed topicPartition which already had an error code, for which the rest of the partitions should exist in InitializeShareGroupStateResult.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2160238278
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +428,70 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ )),
+ new AlterShareGroupOffsetsOptions()
+ .timeoutMs(opts.options.valueOf(opts.timeoutMsOpt).intValue())
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
+ }
+
+ private Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = new HashMap<>();
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+
+ if (opts.options.has(opts.topicOpt)) {
+ Set<String> topics = new HashSet<>(opts.options.valuesOf(opts.topicOpt));
Review Comment:
Good idea. I have changed 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
Re: [PR] KAFKA-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2217941355
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
Review Comment:
Should `DEAD` state be included?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ Map<Uuid, Map<Integer, PartitionErrorData>> topicPartitionErrorsMap = result.getErrors();
Review Comment:
Could you please avoid creating the `topicPartitionErrorsMap` for each loop (`topic`)?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ Map<Uuid, Map<Integer, PartitionErrorData>> topicPartitionErrorsMap = result.getErrors();
+ topic.partitions().forEach(partition -> {
+ if (partition.errorCode() != Errors.NONE.code()) {
+ topicData.partitions().add(partition);
+ return;
+ }
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = topicPartitionErrorsMap.get(topic.topicId()).get(partition.partitionIndex());
Review Comment:
should we check the null for `topicPartitionErrorsMap.get(topic.topicId())`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
Review Comment:
not sure why the topic id is not set for the response
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
+ Map<Uuid, Map<Integer, PartitionErrorData>> topicPartitionErrorsMap = result.getErrors();
+ topic.partitions().forEach(partition -> {
+ if (partition.errorCode() != Errors.NONE.code()) {
+ topicData.partitions().add(partition);
+ return;
+ }
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition partitionData;
+ PartitionErrorData error = topicPartitionErrorsMap.get(topic.topicId()).get(partition.partitionIndex());
+ if (error == null) {
+ partitionData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponsePartition()
Review Comment:
we could reuse the `partition` instead of creating new `AlterShareGroupOffsetsResponsePartition`, right?
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
This behavior is not consistent with normal consumer tool. It should include following tasks.
1. parse the string for "topic" and "topic partition". It means both `--topicc a` and `--topic a:0` are valid
2. parse the `allTopicsOpt`
3. if `allTopicsOpt` is defined, all topics in the group should be reset.
4. if `topicOpt` is defined, all specified tps should be reset
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2164775512
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -3772,7 +3772,19 @@ class KafkaApis(val requestChannel: RequestChannel,
case Some(error) =>
topic.partitions().forEach(partition => responseBuilder.addPartition(topic.topicName(), partition.partitionIndex(), metadataCache.topicNamesToIds(), error.error))
case None =>
- authorizedTopicPartitions.add(topic)
Review Comment:
Thanks for explaining. How about using `authorizedTopicPartitions.add(topic.duplicate())` instead?
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3000025791
Hi @AndrewJSchofield @chia7712
I have updated the PR and resolved several bugs.
Here are some of my test scenarios：
1. When a single shareConsumer subscribes to a single topic, consumes some messages, and then exits and , it can subsequently consume all previously consumed messages again after resetting the offset to ealiest.
2. When a single shareConsumer subscribes to mutiple topics, then excute `bin/kafka-share-groups.sh --reset-offsets --execute --to-earliest --all-topics --group test-group --bootstrap-server localhost:9092 `, the offset of all the topics were reset.
3. When a single shareConsumer subscribes to mutiple topics, consumes some messages, and then exits, it can subsequently consume all previously consumed messages again after resetting the consumption offset.
4. If no topic argument is provided when running `bin/kafka-share-groups.sh --bootstrap-server localhost:9092 --group test-group --reset-offsets --dry-run`, the following exception occurs: Option [reset-offsets] requires one of these options: [all-topics] or [topic].
5. When none of `to-datetime`, `to-earliest` or `to-latest` are specified, the execution failed with an error `Option '[reset-offsets]' requires one of the following scenarios: [[to-datetime], [to-earliest], [to-latest]]`
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2221914472
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
@JimmyWang6 yes, that was a issue in `kafka-consumer-groups.sh` and it was already fixed by @xijiu see #20168 #20064
you could refactor the code to ensure the fixes could be applied on `ShareGroupCommand` too.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2222326661
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,74 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!(GroupState.EMPTY.equals(shareGroupDescription.groupState()) || GroupState.DEAD.equals(shareGroupDescription.groupState()))) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
Review Comment:
What happened if we can't get response from broker?
Should we add try catch or error handler here?
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3074520986
Hi @AndrewJSchofield ,
I have improved the error handling structure, and I'm wondering if we need to distinguish between a non-existent group and a consumer group. Currently, I'm just printing "Share group 'xx' does not exist" for both cases. Here are my test results:
- Attempt to reset offsets for a consumer group
<img width="1399" height="116" alt="image" src="https://github.com/user-attachments/assets/8e5a8f00-4638-45da-a5b9-247c80990c72" />
- Attempt to reset offsets for a non-existent group
<img width="1313" height="103" alt="image" src="https://github.com/user-attachments/assets/09e89680-b50f-45e4-944c-a0f4e9ea5f31" />
- Attempt to reset offsets when there are members in the share group
<img width="1158" height="122" alt="image" src="https://github.com/user-attachments/assets/5e8bd4ab-b13b-430d-9187-96368bef1880" />
- Attempt to reset offsets for non-existent topic
<img width="1415" height="109" alt="image" src="https://github.com/user-attachments/assets/499a4336-c08e-4267-9bf1-603da3fd45fd" />
- Attempt to reset offsets for existing topic which is not subscribed in the share group
<img width="1376" height="85" alt="image" src="https://github.com/user-attachments/assets/90c9267b-9f6b-4876-806f-4bc016b4c057" />
- Success to reset offsets to ealiest
<img width="1428" height="116" alt="image" src="https://github.com/user-attachments/assets/873525f4-1189-462e-b538-e235df0aec68" />
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2208005175
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,67 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() throws ExecutionException, InterruptedException {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ Set<SharePartitionOffsetInformation> partitionOffsets = mapOffsetsToSharePartitionInformation(groupId, offsetsToReset);
+ Map<String, ShareGroupDescription> shareGroups = describeShareGroups(Collections.singleton(groupId));
+ TreeMap<String, Entry<ShareGroupDescription, Collection<SharePartitionOffsetInformation>>> groupOffsets = new TreeMap<>();
+ shareGroups.forEach((id, description) -> groupOffsets.put(id, new SimpleImmutableEntry<>(description, partitionOffsets)));
+ printOffsets(groupOffsets, opts.options.has(opts.verboseOpt));
Review Comment:
Good idea! I will use the same format of `printOffsetsToReset` of consumer group tool.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3125290373
@AndrewJSchofield I've applied the fixes from both PRs and the long delay issue is resolved. However, there might be conflicts with the current changes cc @TaiJuWu.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on PR #19820:
URL: https://github.com/apache/kafka/pull/19820#issuecomment-3134613715
Cheers~
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2220838720
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
@chia7712 @AndrewJSchofield
I have modified the current implementation by reusing offsetsUtils.parseTopicPartitionsToReset(). However, when attempting to pass a non-existent topic-partition (e.g., topicA:999), a TimeoutException is thrown as shown below:
`Caused by: java.util.concurrent.ExecutionException:
org.apache.kafka.common.errors.TimeoutException:
Call(callName=listOffsets(api=METADATA), deadlineMs=1753150292432, tries=73308, nextAllowedTryMs=1753150293436)
timed out at 1753150292436 after 73308 attempt(s)`
This behavior is identical when resetting offsets for a non-existent topic-partition for a consumer group. I think maybe we could create a new issue to address this scenario, as the current implementation does not handle invalid topic-partitions gracefully. I wonder what you 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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2221352615
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -709,15 +709,47 @@ CompletableFuture<AlterShareGroupOffsetsResponseData> persisterInitialize(
handlePersisterInitializeResponse(request.groupTopicPartitionData().groupId(), result, new ShareGroupHeartbeatResponseData());
return response;
} else {
- //TODO build new AlterShareGroupOffsetsResponseData for error response
- return response;
+ return buildErrorResponse(response, result);
}
} else {
return buildErrorResponse(request, response, exp);
}
});
}
+
+ private AlterShareGroupOffsetsResponseData buildErrorResponse(AlterShareGroupOffsetsResponseData response, InitializeShareGroupStateResult result) {
+ AlterShareGroupOffsetsResponseData data = new AlterShareGroupOffsetsResponseData();
+ data.setResponses(
+ new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopicCollection(response.responses().stream()
+ .map(topic -> {
+ AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic topicData = new AlterShareGroupOffsetsResponseData.AlterShareGroupOffsetsResponseTopic()
+ .setTopicName(topic.topicName());
Review Comment:
Sorry, I accidentally missed this issue earlier, but it has now been fixed. Thanks for reminding me!
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2222186794
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
Noted.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2218503657
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
Yes, this is a good point. You cannot delete the offset for individual topic-partitions. You can reset for individual topic-partitions.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "JimmyWang6 (via GitHub)" <gi...@apache.org>.
JimmyWang6 commented on code in PR #19820:
URL: https://github.com/apache/kafka/pull/19820#discussion_r2227559506
##########
tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java:
##########
@@ -366,6 +378,75 @@ Entry<Throwable, Map<String, Throwable>> sendDeleteShareGroupOffsetsRequest(Stri
return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
}
+ void resetOffsets() {
+ String groupId = opts.options.valueOf(opts.groupOpt);
+ try {
+ ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
+ if (!GroupState.EMPTY.equals(shareGroupDescription.groupState())) {
+ CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
+ }
+ Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId);
+ if (offsetsToReset == null) {
+ return;
+ }
+ boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
+ if (!dryRun) {
+ adminClient.alterShareGroupOffsets(groupId,
+ offsetsToReset.entrySet().stream()
+ .collect(Collectors.toMap(
+ Entry::getKey, entry -> entry.getValue().offset()
+ ))
+ ).all().get();
+ }
+ OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
+ } catch (InterruptedException ie) {
+ throw new RuntimeException(ie);
+ } catch (ExecutionException ee) {
+ Throwable cause = ee.getCause();
+ if (cause instanceof KafkaException) {
+ CommandLineUtils.printErrorAndExit(cause.getMessage());
+ } else {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
+
+ protected Map<TopicPartition, OffsetAndMetadata> prepareOffsetsToReset(String groupId) throws ExecutionException, InterruptedException {
+ Map<String, ListShareGroupOffsetsSpec> groupSpecs = Map.of(groupId, new ListShareGroupOffsetsSpec());
+ Map<TopicPartition, OffsetAndMetadata> offsetsByTopicPartitions = adminClient.listShareGroupOffsets(groupSpecs).all().get().get(groupId);
+ Set<String> existsTopics = adminClient.listTopics().names().get();
+
+ if (opts.options.has(opts.topicOpt)) {
Review Comment:
I need to move `checkAllTopicPartitionsValid`, `filterNonExistentPartitions` and `filterNoneLeaderPartitions` method to OffsetUtils to reuse the verification logic for consumer groups. I just noticed there's a remaining issue with that (#20168). I think I could apply the fixes for shared groups after that issue is resolved. Please help to take another look if you get chance.
--
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-16717 [3/N]: Add AdminClient.alterShareGroupOffsets [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #19820:
URL: https://github.com/apache/kafka/pull/19820
--
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