dajac commented on code in PR #15150:
URL: https://github.com/apache/kafka/pull/15150#discussion_r1492178089
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -20,78 +20,225 @@
import kafka.admin.ConsumerGroupCommand;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.common.GroupType;
import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.MethodSource;
import java.util.Arrays;
import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashSet;
import java.util.Objects;
import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
+import static
org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testListConsumerGroups(String quorum) throws Exception {
+ @ParameterizedTest(name =
TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES)
+ @MethodSource("getTestQuorumAndGroupProtocolParametersAll")
+ public void testListConsumerGroupsWithoutFilters(String quorum, String
groupProtocol) throws Exception {
String simpleGroup = "simple-group";
+
+ createOffsetsTopic(listenerName(), new Properties());
+
addSimpleGroupExecutor(simpleGroup);
addConsumerGroupExecutor(1);
+ addConsumerGroupExecutor(1, PROTOCOL_GROUP, groupProtocol);
String[] cgcArgs = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list"};
ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
- scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP,
simpleGroup));
+
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP,
simpleGroup, PROTOCOL_GROUP));
final AtomicReference<scala.collection.Set> foundGroups = new
AtomicReference<>();
+
TestUtils.waitForCondition(() -> {
foundGroups.set(service.listConsumerGroups().toSet());
return Objects.equals(expectedGroups, foundGroups.get());
}, "Expected --list to show groups " + expectedGroups + ", but found "
+ foundGroups.get() + ".");
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testListWithUnrecognizedNewConsumerOption() {
+ @Test
+ public void testListWithUnrecognizedNewConsumerOption() throws Exception {
String[] cgcArgs = new String[]{"--new-consumer",
"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
assertThrows(OptionException.class, () ->
getConsumerGroupService(cgcArgs));
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testListConsumerGroupsWithStates() throws Exception {
+ @ParameterizedTest(name =
TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES)
+ @MethodSource("getTestQuorumAndGroupProtocolParametersAll")
+ public void testListConsumerGroupsWithStates(String quorum, String
groupProtocol) throws Exception {
Review Comment:
`groupProtocol` is never used in this test and we run it with all consumer
types. I suppose that we either want to use `groupProtocol` or only run it with
the classic type. If we do the latter, do we have a test listing new consumer
groups too?
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -143,22 +312,134 @@ public void testListGroupCommand(String quorum) throws
Exception {
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"};
+ String[] cgcArgs3 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type"};
TestUtils.waitForCondition(() -> {
out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
ConsumerGroupCommand.main(cgcArgs3);
return null;
}));
+ return out.get().contains("TYPE") && !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[] cgcArgs4 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "--type"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ }));
+ return out.get().contains("TYPE") && 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[] cgcArgs5 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "Stable"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs5);
+ 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"};
+ String[] cgcArgs6 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "stable"};
TestUtils.waitForCondition(() -> {
out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
- ConsumerGroupCommand.main(cgcArgs4);
+ ConsumerGroupCommand.main(cgcArgs6);
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[] cgcArgs7 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "Classic"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs7);
+ return null;
+ }));
+ return out.get().contains("TYPE") && out.get().contains("Classic")
&& !out.get().contains("STATE") &&
+ out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + GROUP + " and the header, but found " +
out.get());
+
+ String[] cgcArgs8 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "classic"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs8);
+ return null;
+ }));
+ return out.get().contains("TYPE") && out.get().contains("Classic")
&& !out.get().contains("STATE") &&
+ out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + GROUP + " and the header, but found " +
out.get());
Review Comment:
I understand that you follow the pattern in place but I find all those tests
really hard to read/follow. I took a stab at improving them:
https://github.com/apache/kafka/pull/15382. Please let me know what you think.
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -143,22 +312,134 @@ public void testListGroupCommand(String quorum) throws
Exception {
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"};
+ String[] cgcArgs3 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type"};
TestUtils.waitForCondition(() -> {
out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
ConsumerGroupCommand.main(cgcArgs3);
return null;
}));
+ return out.get().contains("TYPE") && !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[] cgcArgs4 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "--type"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ }));
+ return out.get().contains("TYPE") && 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[] cgcArgs5 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "Stable"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs5);
+ 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"};
+ String[] cgcArgs6 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "stable"};
TestUtils.waitForCondition(() -> {
out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
- ConsumerGroupCommand.main(cgcArgs4);
+ ConsumerGroupCommand.main(cgcArgs6);
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[] cgcArgs7 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "Classic"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs7);
+ return null;
+ }));
+ return out.get().contains("TYPE") && out.get().contains("Classic")
&& !out.get().contains("STATE") &&
+ out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + GROUP + " and the header, but found " +
out.get());
+
+ String[] cgcArgs8 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "classic"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs8);
+ return null;
+ }));
+ return out.get().contains("TYPE") && out.get().contains("Classic")
&& !out.get().contains("STATE") &&
+ out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + GROUP + " and the header, but found " +
out.get());
+ }
+
+ @ParameterizedTest(name =
TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES)
+
@MethodSource("getTestQuorumAndGroupProtocolParametersConsumerGroupProtocolOnly")
+ public void testListGroupCommandConsumerProtocol(String quorum, String
groupProtocol) throws Exception {
+ String simpleGroup = "simple-group";
+
+ createOffsetsTopic(listenerName(), new Properties());
+
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1, PROTOCOL_GROUP, groupProtocol);
+
+ 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("TYPE") && !out.get().contains("STATE")
&&
+ out.get().contains(simpleGroup) &&
out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + simpleGroup + ", " + PROTOCOL_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("TYPE")
&& out.get().contains(simpleGroup) && out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + simpleGroup + ", " + PROTOCOL_GROUP + " and
state header, but found " + out.get());
+
+ String[] cgcArgs3 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs3);
+ return null;
+ }));
+ return out.get().contains("TYPE") &&
out.get().contains("Consumer") && !out.get().contains("STATE") &&
+ out.get().contains(PROTOCOL_GROUP) &&
out.get().contains(simpleGroup);
+ }, "Expected to find " + simpleGroup + ", " + PROTOCOL_GROUP + " and
type header, but found " + out.get());
+
+ String[] cgcArgs4 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "consumer"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ }));
+ return out.get().contains("TYPE") &&
out.get().contains("Consumer") && !out.get().contains("STATE") &&
out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + PROTOCOL_GROUP + " with type header and
Consumer type, but found " + out.get());
+
+ String[] cgcArgs5 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "consumer", "--state",
"Stable"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs5);
+ return null;
+ }));
+ return out.get().contains("TYPE") &&
out.get().contains("Consumer") && out.get().contains("STATE") &&
+ out.get().contains("Stable") &&
out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + PROTOCOL_GROUP + " with type and state
header, but found " + out.get());
+ }
+
+ private void assertGroupListing(
Review Comment:
nit: This one could be static.
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -20,78 +20,225 @@
import kafka.admin.ConsumerGroupCommand;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.common.ConsumerGroupState;
+import org.apache.kafka.common.GroupType;
import org.apache.kafka.test.TestUtils;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.MethodSource;
import java.util.Arrays;
import java.util.Collections;
+import java.util.EnumSet;
+import java.util.HashSet;
import java.util.Objects;
import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
+import static
org.apache.kafka.tools.ToolsTestUtils.TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class ListConsumerGroupTest extends ConsumerGroupCommandTest {
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testListConsumerGroups(String quorum) throws Exception {
+ @ParameterizedTest(name =
TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES)
+ @MethodSource("getTestQuorumAndGroupProtocolParametersAll")
+ public void testListConsumerGroupsWithoutFilters(String quorum, String
groupProtocol) throws Exception {
String simpleGroup = "simple-group";
+
+ createOffsetsTopic(listenerName(), new Properties());
+
addSimpleGroupExecutor(simpleGroup);
addConsumerGroupExecutor(1);
+ addConsumerGroupExecutor(1, PROTOCOL_GROUP, groupProtocol);
String[] cgcArgs = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list"};
ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(cgcArgs);
- scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP,
simpleGroup));
+
+ scala.collection.Set<String> expectedGroups = set(Arrays.asList(GROUP,
simpleGroup, PROTOCOL_GROUP));
final AtomicReference<scala.collection.Set> foundGroups = new
AtomicReference<>();
+
TestUtils.waitForCondition(() -> {
foundGroups.set(service.listConsumerGroups().toSet());
return Objects.equals(expectedGroups, foundGroups.get());
}, "Expected --list to show groups " + expectedGroups + ", but found "
+ foundGroups.get() + ".");
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testListWithUnrecognizedNewConsumerOption() {
+ @Test
+ public void testListWithUnrecognizedNewConsumerOption() throws Exception {
String[] cgcArgs = new String[]{"--new-consumer",
"--bootstrap-server", bootstrapServers(listenerName()), "--list"};
assertThrows(OptionException.class, () ->
getConsumerGroupService(cgcArgs));
}
- @ParameterizedTest
- @ValueSource(strings = {"zk", "kraft"})
- public void testListConsumerGroupsWithStates() throws Exception {
+ @ParameterizedTest(name =
TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES)
+ @MethodSource("getTestQuorumAndGroupProtocolParametersAll")
+ public void testListConsumerGroupsWithStates(String quorum, String
groupProtocol) throws Exception {
String simpleGroup = "simple-group";
+
+ createOffsetsTopic(listenerName(), new Properties());
Review Comment:
Yes, I would keep them.
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -143,22 +312,134 @@ public void testListGroupCommand(String quorum) throws
Exception {
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"};
+ String[] cgcArgs3 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type"};
TestUtils.waitForCondition(() -> {
out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
ConsumerGroupCommand.main(cgcArgs3);
return null;
}));
+ return out.get().contains("TYPE") && !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[] cgcArgs4 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "--type"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ }));
+ return out.get().contains("TYPE") && 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[] cgcArgs5 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "Stable"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs5);
+ 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"};
+ String[] cgcArgs6 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--state", "stable"};
TestUtils.waitForCondition(() -> {
out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
- ConsumerGroupCommand.main(cgcArgs4);
+ ConsumerGroupCommand.main(cgcArgs6);
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[] cgcArgs7 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "Classic"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs7);
+ return null;
+ }));
+ return out.get().contains("TYPE") && out.get().contains("Classic")
&& !out.get().contains("STATE") &&
+ out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + GROUP + " and the header, but found " +
out.get());
+
+ String[] cgcArgs8 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "classic"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs8);
+ return null;
+ }));
+ return out.get().contains("TYPE") && out.get().contains("Classic")
&& !out.get().contains("STATE") &&
+ out.get().contains(simpleGroup) && out.get().contains(GROUP);
+ }, "Expected to find " + GROUP + " and the header, but found " +
out.get());
+ }
+
+ @ParameterizedTest(name =
TEST_WITH_PARAMETERIZED_QUORUM_AND_GROUP_PROTOCOL_NAMES)
+
@MethodSource("getTestQuorumAndGroupProtocolParametersConsumerGroupProtocolOnly")
+ public void testListGroupCommandConsumerProtocol(String quorum, String
groupProtocol) throws Exception {
+ String simpleGroup = "simple-group";
+
+ createOffsetsTopic(listenerName(), new Properties());
+
+ addSimpleGroupExecutor(simpleGroup);
+ addConsumerGroupExecutor(1, PROTOCOL_GROUP, groupProtocol);
+
+ 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("TYPE") && !out.get().contains("STATE")
&&
+ out.get().contains(simpleGroup) &&
out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + simpleGroup + ", " + PROTOCOL_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("TYPE")
&& out.get().contains(simpleGroup) && out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + simpleGroup + ", " + PROTOCOL_GROUP + " and
state header, but found " + out.get());
+
+ String[] cgcArgs3 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs3);
+ return null;
+ }));
+ return out.get().contains("TYPE") &&
out.get().contains("Consumer") && !out.get().contains("STATE") &&
+ out.get().contains(PROTOCOL_GROUP) &&
out.get().contains(simpleGroup);
+ }, "Expected to find " + simpleGroup + ", " + PROTOCOL_GROUP + " and
type header, but found " + out.get());
+
+ String[] cgcArgs4 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "consumer"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs4);
+ return null;
+ }));
+ return out.get().contains("TYPE") &&
out.get().contains("Consumer") && !out.get().contains("STATE") &&
out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + PROTOCOL_GROUP + " with type header and
Consumer type, but found " + out.get());
+
+ String[] cgcArgs5 = new String[]{"--bootstrap-server",
bootstrapServers(listenerName()), "--list", "--type", "consumer", "--state",
"Stable"};
+ TestUtils.waitForCondition(() -> {
+ out.set(kafka.utils.TestUtils.grabConsoleOutput(() -> {
+ ConsumerGroupCommand.main(cgcArgs5);
+ return null;
+ }));
+ return out.get().contains("TYPE") &&
out.get().contains("Consumer") && out.get().contains("STATE") &&
+ out.get().contains("Stable") &&
out.get().contains(PROTOCOL_GROUP);
+ }, "Expected to find " + PROTOCOL_GROUP + " with type and state
header, but found " + out.get());
+ }
+
+ private void assertGroupListing(
+ Set<GroupType> typeFilterSet,
+ Set<ConsumerGroupState> stateFilterSet,
+ Set<ConsumerGroupListing> expectedListing,
+ ConsumerGroupCommand.ConsumerGroupService service
Review Comment:
nit: Let's put the service first.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
