You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "aliehsaeedii (via GitHub)" <gi...@apache.org> on 2024/12/17 14:30:04 UTC
[PR] Implement kafka-streams-groups.sh --describe [kafka]
aliehsaeedii opened a new pull request, #18231:
URL: https://github.com/apache/kafka/pull/18231
Implement --describe and its options: (--state, --offset, --members and the combination of them with --verbose)
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1912747227
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -205,23 +206,23 @@ private void printMembers(StreamsGroupDescription description, boolean verbose)
System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
for (StreamsGroupMemberDescription member : members) {
System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
- getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ member.assignment().activeTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
Review Comment:
> How are you testing these changes?
Do you mean testing the output of `print` methods? The counterpart classes (`ShareGroup` etc) dont test it. Do you think is it needed?
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890586538
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
Review Comment:
> Looks like printing offsets is the default - would it make sense to have it in an `else if` and the `else` would print something like `Unknown option`?
I think we should manage such things (`Unknown Ooptions`) in `StreamsGroupCommandOptions`.
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1922412283
##########
tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandUnitTest.java:
##########
@@ -163,6 +251,7 @@ public void testGroupStatesFromString() {
assertThrows(IllegalArgumentException.class, () -> StreamsGroupCommand.groupStatesFromString(" , ,"));
}
+
Review Comment:
nit: remove
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890591797
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
Review Comment:
Now with `--describe --xyz` where `xyz` is unknown, the command's output is the same as `--describe` or `describe --offsets`
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890623199
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ if (descriptionMap.containsKey(group)) {
+ return descriptionMap.get(group);
+ }
+ return null;
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
Review Comment:
Oh, Got 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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1912750089
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -205,23 +206,23 @@ private void printMembers(StreamsGroupDescription description, boolean verbose)
System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
for (StreamsGroupMemberDescription member : members) {
System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
- getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ member.assignment().activeTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
+ member.assignment().standbyTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
+ member.assignment().warmupTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")));
}
} else {
- String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ String fmt = "%" + -groupLen + "s %-15s% %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ "%s %s %s %s %s %s\n\n";
- System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "TOPOLOGY-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
Review Comment:
The header is the string `TARGET-ASSIGNMENT-EPOCH` and also have to fit the column.
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890661710
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
Review Comment:
@mjsax do you agree with this way of printing assignments?
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1892136315
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s %s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS", "TARGET-ACTIVE-TASKS", "TARGET-STANDBY-TASKS", "TARGET-WARMUP-TASKS");
Review Comment:
Can we also show the Topology Epoch?
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s %s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS", "TARGET-ACTIVE-TASKS", "TARGET-STANDBY-TASKS", "TARGET-WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), description.targetAssignmentEpoch(), member.memberId(), member.isClassic() ? "Classic Protocol" : "New Protocol", member.memberEpoch(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
Review Comment:
This does not make much sense to me. Why are converting tasks to topic partitions to print them?
Maybe just implement a method taskIdsToString and use it everywhere.
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s %s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS", "TARGET-ACTIVE-TASKS", "TARGET-STANDBY-TASKS", "TARGET-WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), description.targetAssignmentEpoch(), member.memberId(), member.isClassic() ? "Classic Protocol" : "New Protocol", member.memberEpoch(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ }
+ }
+ }
+
+ private void printStates(StreamsGroupDescription description, boolean verbose) {
+ maybePrintEmptyGroupState(description.groupId(), description.groupState(), 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());
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %-15s %s\n";
+ System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE", "#MEMBERS");
+ System.out.printf(fmt, description.groupId(), coordinator, description.groupState().toString(), description.members().size());
+ } else {
+ String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %-15s %-15s %-15s %s\n";
+ System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE", "GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS");
+ System.out.printf(fmt, description.groupId(), coordinator, description.groupState().toString(), description.groupEpoch(), description.targetAssignmentEpoch(), description.members().size());
+ }
+ }
+
+ private void printOffsets(StreamsGroupDescription description, boolean verbose) throws ExecutionException, InterruptedException {
+ Map<TopicPartition, Long> offsets = getOffsets(description.members());
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), offsets.size())) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxTopicLen = 15;
+ for (TopicPartition topicPartition : offsets.keySet()) {
+ maxTopicLen = Math.max(maxTopicLen, topicPartition.topic().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + (-groupLen) + "s %" + (-maxTopicLen) + "s %-10s %s\n";
+ System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "OFFSET-LAG");
+ for (Map.Entry<TopicPartition, Long> offset : offsets.entrySet()) {
+ System.out.printf(fmt, description.groupId(), offset.getKey().topic(), offset.getKey().partition(), offset.getValue());
+ }
+ } else {
+ String fmt = "%" + (-groupLen) + "s %" + (-maxTopicLen) + "s %-10s %-15s %s\n";
+ System.out.printf(fmt, "GROUP", "TOPIC", "PARTITION", "LEADER-EPOCH", "OFFSET-LAG");
+ for (Map.Entry<TopicPartition, Long> offset : offsets.entrySet()) {
+ System.out.printf(fmt, description.groupId(), offset.getKey().topic(), offset.getKey().partition(), "", offset.getValue());
+ }
+ }
+ }
+ }
+
+ Map<TopicPartition, Long> getOffsets(Collection<StreamsGroupMemberDescription> members) throws ExecutionException, InterruptedException {
+ Set<TopicPartition> allTp = new HashSet<>();
+ for (StreamsGroupMemberDescription memberDescription : members) {
+ allTp.addAll(getTopicPartitions(memberDescription.assignment().activeTasks()));
Review Comment:
A task is not a topic partition!
You need to get the topology from the StreamsGroupDescribeResponse to map tasks to source topics.
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s %s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS", "TARGET-ACTIVE-TASKS", "TARGET-STANDBY-TASKS", "TARGET-WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), description.targetAssignmentEpoch(), member.memberId(), member.isClassic() ? "Classic Protocol" : "New Protocol", member.memberEpoch(), member.processId(), member.clientId(),
Review Comment:
"New Protocol" is not really clear, and will be annoying if we ever introduce another protocol. Maybe just use the group type enums, "classic" for the old protocol and "streams" for the new protocol.
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1912749627
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -205,23 +206,23 @@ private void printMembers(StreamsGroupDescription description, boolean verbose)
System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
for (StreamsGroupMemberDescription member : members) {
System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
- getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ member.assignment().activeTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
Review Comment:
Not necessarily a unit tests. But I think it would be good to actually run the tool, even if we do not have an integration test yet. I think if we'd run it, we'd spot this kind of formatting problems easily.
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru merged PR #18231:
URL: https://github.com/apache/kafka/pull/18231
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890479721
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
Review Comment:
Looks like printing offsets is the default - would it make sense to have it in an `else if` and the `else` would print something like `Unknown option`?
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ if (descriptionMap.containsKey(group)) {
+ return descriptionMap.get(group);
+ }
+ return null;
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
Review Comment:
I thought all the placeholders need to be `%s`
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ if (descriptionMap.containsKey(group)) {
+ return descriptionMap.get(group);
+ }
+ return null;
Review Comment:
```suggestion
return descriptionMap.get(group);
```
Since this method can return `null` can be simplified to one line
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ if (descriptionMap.containsKey(group)) {
+ return descriptionMap.get(group);
+ }
+ return null;
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
+
+ if (!verbose) {
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS", "TARGET-ACTIVE-TASKS", "TARGET-STANDBY-TASKS", "TARGET-WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), description.targetAssignmentEpoch(), member.memberId(), member.isClassic() ? "Classic Protocol" : "New Protocol", member.memberEpoch(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ }
+ }
+ }
+
+ private void printStates(StreamsGroupDescription description, boolean verbose) {
+ maybePrintEmptyGroupState(description.groupId(), description.groupState(), 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());
+ String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %-15s %s\n";
+
+ if (!verbose) {
+ System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE", "#MEMBERS");
+ System.out.printf(fmt, description.groupId(), coordinator, description.groupState().toString(), description.members().size());
+ } else {
+ System.out.printf(fmt, "GROUP", "COORDINATOR (ID)", "STATE", "GROUP-EPOCH", "TARGET-ASSIGNMENT-EPOCH", "#MEMBERS");
+ System.out.printf(fmt, description.groupId(), coordinator, description.groupState().toString(), description.groupEpoch(), description.targetAssignmentEpoch(), description.members().size());
+ }
+ }
+
+ private void printOffsets(StreamsGroupDescription description, boolean verbose) throws ExecutionException, InterruptedException {
+ Map<TopicPartition, Long> offsets = getOffsets(description.members());
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), offsets.size())) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxTopicLen = 15;
+ for (TopicPartition topicPartition : offsets.keySet()) {
+ maxTopicLen = Math.max(maxTopicLen, topicPartition.topic().length());
+ }
Review Comment:
And here
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ if (descriptionMap.containsKey(group)) {
+ return descriptionMap.get(group);
+ }
+ return null;
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
+
+ if (!verbose) {
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS", "TARGET-ACTIVE-TASKS", "TARGET-STANDBY-TASKS", "TARGET-WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), description.targetAssignmentEpoch(), member.memberId(), member.isClassic() ? "Classic Protocol" : "New Protocol", member.memberEpoch(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.targetAssignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ }
+ }
+ }
+
+ private void printStates(StreamsGroupDescription description, boolean verbose) {
+ maybePrintEmptyGroupState(description.groupId(), description.groupState(), 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());
+ String fmt = "%" + -groupLen + "s %" + -coordinatorLen + "s %-15s %s\n";
Review Comment:
Same 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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890661710
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
Review Comment:
@mjsax do you agree with this way of printing assignments?
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1912697855
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -205,23 +206,23 @@ private void printMembers(StreamsGroupDescription description, boolean verbose)
System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
for (StreamsGroupMemberDescription member : members) {
System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
- getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ member.assignment().activeTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
+ member.assignment().standbyTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
+ member.assignment().warmupTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")));
}
} else {
- String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ String fmt = "%" + -groupLen + "s %-15s% %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ "%s %s %s %s %s %s\n\n";
- System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "TOPOLOGY-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
Review Comment:
> Your column seems to be 15 characters, but TARGET-ASSIGNMENT-EPOCH is >20 character. WOn't this look misaligned?
How? I assumed it's an `int`.
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "lucasbru (via GitHub)" <gi...@apache.org>.
lucasbru commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1910610974
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -205,23 +206,23 @@ private void printMembers(StreamsGroupDescription description, boolean verbose)
System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
for (StreamsGroupMemberDescription member : members) {
System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
- getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ member.assignment().activeTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
Review Comment:
I wonder how readable this will be. I think it will look something like this:
```
ACTIVE-TASKS STANDBY-TASKS WARMUP-TASKS
subtopology1:[1, 2, 3],subtopology2:[1] subtopology1:[1] subtopology2:[9]
```
Have you thought about how to make this more readable?
How are you testing these changes?
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -205,23 +206,23 @@ private void printMembers(StreamsGroupDescription description, boolean verbose)
System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
for (StreamsGroupMemberDescription member : members) {
System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
- getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
- getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ member.assignment().activeTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
+ member.assignment().standbyTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")),
+ member.assignment().warmupTasks().stream().map(taskId -> taskId.subtopologyId() + ":" + taskId.partitions()).collect(Collectors.joining(",")));
}
} else {
- String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ String fmt = "%" + -groupLen + "s %-15s% %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ "%s %s %s %s %s %s\n\n";
- System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
+ System.out.printf(fmt, "GROUP", "TARGET-ASSIGNMENT-EPOCH", "TOPOLOGY-EPOCH", "MEMBER", "MEMBER-PROTOCOL", "MEMBER-EPOCH", "PROCESS", "CLIENT",
Review Comment:
Your column seems to be 15 characters, but TARGET-ASSIGNMENT-EPOCH is >20 character. WOn't this look misaligned?
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s %s %s %s\n\n";
Review Comment:
Same here. If we cannot align this data in columns, we probably need to represent it in one line, but then we cannot differentiate between the task types (active, standby, warm-up) using column headers, but in the line.
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "akshayw1 (via GitHub)" <gi...@apache.org>.
akshayw1 commented on PR #18231:
URL: https://github.com/apache/kafka/pull/18231#issuecomment-2549697172
Please assign
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890587650
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,156 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ if (descriptionMap.containsKey(group)) {
+ return descriptionMap.get(group);
+ }
+ return null;
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s %s\n";
Review Comment:
> I thought all the placeholders need to be `%s`
What's the problem with `%-20s`?
--
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] Implement kafka-streams-groups.sh --describe [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #18231:
URL: https://github.com/apache/kafka/pull/18231#discussion_r1890662743
##########
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java:
##########
@@ -156,6 +167,158 @@ private void printGroupInfo(List<GroupListing> groups) {
}
}
+ public void describeGroups() throws ExecutionException, InterruptedException {
+ String group = opts.options.valueOf(opts.groupOpt);
+ StreamsGroupDescription description = getDescribeGroup(group);
+ if (description == null)
+ return;
+ boolean verbose = opts.options.has(opts.verboseOpt);
+ if (opts.options.has(opts.membersOpt)) {
+ printMembers(description, verbose);
+ } else if (opts.options.has(opts.stateOpt)) {
+ printStates(description, verbose);
+ } else {
+ printOffsets(description, verbose);
+ }
+ }
+
+ StreamsGroupDescription getDescribeGroup(String group) throws ExecutionException, InterruptedException {
+ DescribeStreamsGroupsResult result = adminClient.describeStreamsGroups(List.of(group));
+ Map<String, StreamsGroupDescription> descriptionMap = result.all().get();
+ return descriptionMap.get(group);
+ }
+
+ private void printMembers(StreamsGroupDescription description, boolean verbose) {
+ int groupLen = Math.max(15, description.groupId().length());
+ int maxMemberIdLen = 15, maxHostLen = 15, maxClientIdLen = 15;
+ Collection<StreamsGroupMemberDescription> members = description.members();
+ if (maybePrintEmptyGroupState(description.groupId(), description.groupState(), description.members().size())) {
+ for (StreamsGroupMemberDescription member : members) {
+ maxMemberIdLen = Math.max(maxMemberIdLen, member.memberId().length());
+ maxHostLen = Math.max(maxHostLen, member.processId().length());
+ maxClientIdLen = Math.max(maxClientIdLen, member.clientId().length());
+ }
+
+ if (!verbose) {
+ String fmt = "%" + -groupLen + "s %" + -maxMemberIdLen + "s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s\n\n";
+ System.out.printf(fmt, "GROUP", "MEMBER", "PROCESS", "CLIENT-ID", "ACTIVE-TASKS", "STANDBY-TASKS", "WARMUP-TASKS");
+ for (StreamsGroupMemberDescription member : members) {
+ System.out.printf(fmt, description.groupId(), member.memberId(), member.processId(), member.clientId(),
+ getTopicPartitions(member.assignment().activeTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().standbyTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")),
+ getTopicPartitions(member.assignment().warmupTasks()).stream().map(tp -> tp.topic() + ":" + tp.partition()).collect(Collectors.joining(",")));
+ }
+ } else {
+ String fmt = "%" + -groupLen + "s %-15s%" + -maxMemberIdLen + "s %s %15s %" + -maxHostLen + "s %" + -maxClientIdLen + "s\n"
+ + "%s %s %s %s %s %s\n\n";
Review Comment:
The same question here @mjsax
--
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