You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2024/02/17 08:07:56 UTC
(kafka) branch trunk updated: MINOR: Improve ListConsumerGroupTest.testListGroupCommand (#15382)
This is an automated email from the ASF dual-hosted git repository.
dajac 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 e247bd03afe MINOR: Improve ListConsumerGroupTest.testListGroupCommand (#15382)
e247bd03afe is described below
commit e247bd03afe66d61426a9029220d06438dede3dc
Author: David Jacot <dj...@confluent.io>
AuthorDate: Sat Feb 17 00:07:50 2024 -0800
MINOR: Improve ListConsumerGroupTest.testListGroupCommand (#15382)
While reviewing https://github.com/apache/kafka/pull/15150, I found that our tests verifying the console output are really hard to read. Here is my proposal to make it better.
Reviewers: Ritika Reddy <rr...@confluent.io>, Justine Olshan <jo...@confluent.io>
---
.../consumer/group/ListConsumerGroupTest.java | 121 +++++++++++++++------
1 file changed, 86 insertions(+), 35 deletions(-)
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java
index 6648cfe2a24..894f00df5e7 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java
@@ -26,10 +26,15 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import static org.apache.kafka.common.utils.Utils.mkSet;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -123,42 +128,88 @@ public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
String simpleGroup = "simple-group";
addSimpleGroupExecutor(simpleGroup);
addConsumerGroupExecutor(1);
- final AtomicReference<String> out = new AtomicReference<>("");
- String[] cgcArgs1 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
- TestUtils.waitForCondition(() -> {
- out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
- ConsumerGroupCommand.main(cgcArgs1);
- return null;
- }));
- return !out.get().contains("STATE") && out.get().contains(simpleGroup) && out.get().contains(GROUP);
- }, "Expected to find " + simpleGroup + ", " + GROUP + " and no header, but found " + out.get());
-
- String[] cgcArgs2 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"};
- TestUtils.waitForCondition(() -> {
- out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
- ConsumerGroupCommand.main(cgcArgs2);
- return null;
- }));
- return out.get().contains("STATE") && out.get().contains(simpleGroup) && out.get().contains(GROUP);
- }, "Expected to find " + simpleGroup + ", " + GROUP + " and the header, but found " + out.get());
-
- String[] cgcArgs3 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "Stable"};
- TestUtils.waitForCondition(() -> {
- out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
- ConsumerGroupCommand.main(cgcArgs3);
- return null;
- }));
- return out.get().contains("STATE") && out.get().contains(GROUP) && out.get().contains("Stable");
- }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out.get());
-
- String[] cgcArgs4 = new String[]{"--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "stable"};
+ validateListOutput(
+ Arrays.asList("--bootstrap-server", bootstrapServers(listenerName()), "--list"),
+ Collections.emptyList(),
+ mkSet(
+ Collections.singletonList(GROUP),
+ Collections.singletonList(simpleGroup)
+ )
+ );
+
+ validateListOutput(
+ Arrays.asList("--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state"),
+ Arrays.asList("GROUP", "STATE"),
+ mkSet(
+ Arrays.asList(GROUP, "Stable"),
+ Arrays.asList(simpleGroup, "Empty")
+ )
+ );
+
+ validateListOutput(
+ Arrays.asList("--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "Stable"),
+ Arrays.asList("GROUP", "STATE"),
+ mkSet(
+ Arrays.asList(GROUP, "Stable")
+ )
+ );
+
+ validateListOutput(
+ Arrays.asList("--bootstrap-server", bootstrapServers(listenerName()), "--list", "--state", "stable"),
+ Arrays.asList("GROUP", "STATE"),
+ mkSet(
+ Arrays.asList(GROUP, "Stable")
+ )
+ );
+ }
+
+ /**
+ * Validates that the output of the list command corresponds to the expected values.
+ *
+ * @param args The arguments for the command line tool.
+ * @param expectedHeader The expected header as a list of strings; or an empty list
+ * if a header is not expected.
+ * @param expectedRows The expected rows as a set of list of columns.
+ * @throws InterruptedException
+ */
+ private static void validateListOutput(
+ List<String> args,
+ List<String> expectedHeader,
+ Set<List<String>> expectedRows
+ ) throws InterruptedException {
+ final AtomicReference<String> out = new AtomicReference<>("");
TestUtils.waitForCondition(() -> {
- out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
- ConsumerGroupCommand.main(cgcArgs4);
- return null;
- }));
- return out.get().contains("STATE") && out.get().contains(GROUP) && out.get().contains("Stable");
- }, "Expected to find " + GROUP + " in state Stable and the header, but found " + out.get());
+ String output = runAndGrabConsoleOutput(args);
+ out.set(output);
+
+ int index = 0;
+ String[] lines = output.split("\n");
+
+ // Parse the header if one is expected.
+ if (!expectedHeader.isEmpty()) {
+ if (lines.length == 0) return false;
+ List<String> header = Arrays.stream(lines[index++].split("\\s+")).collect(Collectors.toList());
+ if (!expectedHeader.equals(header)) {
+ return false;
+ }
+ }
+
+ // Parse the groups.
+ Set<List<String>> groups = new HashSet<>();
+ for (; index < lines.length; index++) {
+ groups.add(Arrays.stream(lines[index].split("\\s+")).collect(Collectors.toList()));
+ }
+ return expectedRows.equals(groups);
+ }, () -> String.format("Expected header=%s and groups=%s, but found:%n%s", expectedHeader, expectedRows, out.get()));
+ }
+
+ private static String runAndGrabConsoleOutput(
+ List<String> args
+ ) {
+ return kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(args.toArray(new String[0]));
+ return null;
+ });
}
}