You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ma...@apache.org on 2024/08/20 06:35:14 UTC
(kafka) branch trunk updated: KAFKA-17350: Improve share group describe for empty groups (#16897)
This is an automated email from the ASF dual-hosted git repository.
manikumar pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new f6bfa9406ca KAFKA-17350: Improve share group describe for empty groups (#16897)
f6bfa9406ca is described below
commit f6bfa9406ca88d611e9c9ea0120b16d37e6ce015
Author: Andrew Schofield <as...@confluent.io>
AuthorDate: Tue Aug 20 07:35:08 2024 +0100
KAFKA-17350: Improve share group describe for empty groups (#16897)
When you use kafka-share-groups.sh --describe for an empty group, it prints an empty table consisting of only the table header. kafka-consumer-groups.sh summarises the group status to make the output more informative and only prints the table if it contains more than zero rows.
This PR applies this principle across all of the variants of describing share groups which makes the output much nicer where the output would otherwise be strangely empty.
Reviewers: Manikumar Reddy <ma...@gmail.com>
---
.../tools/consumer/group/ShareGroupCommand.java | 43 ++++++++++++++++------
.../consumer/group/ShareGroupCommandTest.java | 11 ++++++
2 files changed, 42 insertions(+), 12 deletions(-)
diff --git a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java
index c6bdfcfbfc4..4e7dd8a45c5 100644
--- a/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java
+++ b/tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java
@@ -169,7 +169,22 @@ public class ShareGroupCommand {
}
}
- private void describeGroups() throws ExecutionException, InterruptedException {
+ /**
+ * Prints a summary of the state for situations where the group is empty or dead.
+ *
+ * @return Whether the group detail should be printed
+ */
+ public static boolean maybePrintEmptyGroupState(String group, ShareGroupState state, int numRows) {
+ if (state == ShareGroupState.DEAD) {
+ printError("Share group '" + group + "' does not exist.", Optional.empty());
+ } else if (state == ShareGroupState.EMPTY) {
+ System.err.println("\nShare group '" + group + "' has no active members.");
+ }
+
+ return !state.equals(ShareGroupState.DEAD) && numRows > 0;
+ }
+
+ public void describeGroups() throws ExecutionException, InterruptedException {
String group = opts.options.valueOf(opts.groupOpt);
ShareGroupDescription description = getDescribeGroup(group);
if (description == null)
@@ -218,7 +233,7 @@ public class ShareGroupCommand {
private void printOffsets(ShareGroupDescription description) throws ExecutionException, InterruptedException {
Map<TopicPartition, Long> offsets = getOffsets(description.members());
- if (offsets != null && !offsets.isEmpty()) {
+ if (maybePrintEmptyGroupState(description.groupId(), description.state(), offsets.size())) {
String fmt = printOffsetFormat(description, offsets);
System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "OFFSET");
@@ -238,6 +253,8 @@ public class ShareGroupCommand {
}
private void printStates(ShareGroupDescription description) {
+ maybePrintEmptyGroupState(description.groupId(), description.state(), 1);
+
int groupLen = Math.max(15, description.groupId().length());
String coordinator = description.coordinator().host() + ":" + description.coordinator().port() + " (" + description.coordinator().idString() + ")";
int coordinatorLen = Math.max(25, coordinator.length());
@@ -251,17 +268,19 @@ public class ShareGroupCommand {
int groupLen = Math.max(15, description.groupId().length());
int maxConsumerIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
Collection<MemberDescription> members = description.members();
- for (MemberDescription member : members) {
- maxConsumerIdLen = Math.max(maxConsumerIdLen, member.consumerId().length());
- maxHostLen = Math.max(maxHostLen, member.host().length());
- maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
- }
+ if (maybePrintEmptyGroupState(description.groupId(), description.state(), description.members().size())) {
+ for (MemberDescription member : members) {
+ maxConsumerIdLen = Math.max(maxConsumerIdLen, member.consumerId().length());
+ maxHostLen = Math.max(maxHostLen, member.host().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
- String fmt = "%" + -groupLen + "s %" + -maxConsumerIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
- System.out.printf(fmt, "GROUP", "CONSUMER-ID", "HOST", "CLIENT-ID", "ASSIGNMENT");
- for (MemberDescription member : members) {
- System.out.printf(fmt, description.groupId(), member.consumerId(), member.host(), member.clientId(),
- member.assignment().topicPartitions().stream().map(part -> part.topic() + ":" + part.partition()).collect(Collectors.joining(",")));
+ String fmt = "%" + -groupLen + "s %" + -maxConsumerIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
+ System.out.printf(fmt, "GROUP", "CONSUMER-ID", "HOST", "CLIENT-ID", "ASSIGNMENT");
+ for (MemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.consumerId(), member.host(), member.clientId(),
+ member.assignment().topicPartitions().stream().map(part -> part.topic() + ":" + part.partition()).collect(Collectors.joining(",")));
+ }
}
}
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java
index 86345cc75b4..b45902f86e3 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ShareGroupCommandTest.java
@@ -49,7 +49,9 @@ import java.util.Set;
import joptsimple.OptionException;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -129,6 +131,15 @@ public class ShareGroupCommandTest {
service.close();
}
+ @Test
+ public void testPrintEmptyGroupState() {
+ assertFalse(ShareGroupService.maybePrintEmptyGroupState("group", ShareGroupState.EMPTY, 0));
+ assertFalse(ShareGroupService.maybePrintEmptyGroupState("group", ShareGroupState.DEAD, 0));
+ assertFalse(ShareGroupService.maybePrintEmptyGroupState("group", ShareGroupState.STABLE, 0));
+ assertTrue(ShareGroupService.maybePrintEmptyGroupState("group", ShareGroupState.STABLE, 1));
+ assertTrue(ShareGroupService.maybePrintEmptyGroupState("group", ShareGroupState.UNKNOWN, 1));
+ }
+
@Test
public void testListWithUnrecognizedOption() {
String bootstrapServer = "localhost:9092";