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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
