Russole commented on code in PR #23378:
URL: https://github.com/apache/kafka/pull/23378#discussion_r3959310057
##########
tools/src/test/java/org/apache/kafka/tools/consumer/group/ListConsumerGroupTest.java:
##########
@@ -66,78 +61,87 @@
import static
org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
import static
org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_PARTITIONS_CONFIG;
import static
org.apache.kafka.coordinator.group.GroupCoordinatorConfig.OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
-@ClusterTestDefaults(
- types = {Type.CO_KRAFT},
- serverProperties = {
- @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value =
"1"),
- @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG,
value = "1"),
- @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG,
value = "1000"),
- @ClusterConfigProperty(key =
CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500"),
- @ClusterConfigProperty(key =
CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500"),
- }
-)
public class ListConsumerGroupTest {
private static final String TOPIC_PREFIX = "test.topic.";
private static final String TOPIC_PARTITIONS_GROUP_PREFIX =
"test.topic.partitions.group.";
private static final String TOPIC_GROUP_PREFIX = "test.topic.group.";
private static final String PROTOCOL_GROUP_PREFIX = "test.protocol.group.";
- private final ClusterInstance clusterInstance;
-
- ListConsumerGroupTest(ClusterInstance clusterInstance) {
- this.clusterInstance = clusterInstance;
- }
+ private static final String DUMMY_BOOTSTRAP_SERVERS = "localhost:9092";
- private List<GroupProtocol> supportedGroupProtocols() {
+ private static List<GroupProtocol> supportedGroupProtocols(ClusterInstance
clusterInstance) {
return new ArrayList<>(clusterInstance.supportedGroupProtocols());
}
- @ClusterTest
- public void testListConsumerGroupsWithoutFilters() throws Exception {
- for (int i = 0; i < supportedGroupProtocols().size(); i++) {
- GroupProtocol groupProtocol = supportedGroupProtocols().get(i);
+ @ClusterTest(
+ types = {Type.CO_KRAFT},
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG,
value = "1"),
+ @ClusterConfigProperty(key =
OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key =
GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "1000"),
+ @ClusterConfigProperty(key =
CONSUMER_GROUP_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500"),
+ @ClusterConfigProperty(key =
CONSUMER_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "500")
+ }
+ )
+ public void testListConsumerGroupsWithoutFilters(ClusterInstance
clusterInstance) throws Exception {
+ List<GroupProtocol> groupProtocols =
supportedGroupProtocols(clusterInstance);
+ for (int i = 0; i < groupProtocols.size(); i++) {
+ GroupProtocol groupProtocol = groupProtocols.get(i);
String topic = TOPIC_PREFIX + groupProtocol.name;
String protocolGroup = PROTOCOL_GROUP_PREFIX + groupProtocol.name;
String topicGroup = TOPIC_GROUP_PREFIX + i;
String topicPartitionsGroup = TOPIC_PARTITIONS_GROUP_PREFIX + i;
- createTopic(topic);
+ clusterInstance.createTopic(topic, 1, (short) 1);
- try (AutoCloseable topicPartitionsConsumerGroupExecutor =
consumerGroupClosable(topicPartitionsGroup, Set.of(new TopicPartition(topic,
0)));
- AutoCloseable topicConsumerGroupExecutor =
consumerGroupClosable(GroupProtocol.CLASSIC, topicGroup, topic);
- AutoCloseable protocolConsumerGroupExecutor =
consumerGroupClosable(groupProtocol, protocolGroup, topic);
+ try (AutoCloseable topicPartitionsConsumerGroupExecutor =
consumerGroupClosable(clusterInstance, topicPartitionsGroup, Set.of(new
TopicPartition(topic, 0)));
+ AutoCloseable topicConsumerGroupExecutor =
consumerGroupClosable(clusterInstance, GroupProtocol.CLASSIC, topicGroup,
topic);
+ AutoCloseable protocolConsumerGroupExecutor =
consumerGroupClosable(clusterInstance, groupProtocol, protocolGroup, topic);
ConsumerGroupCommand.ConsumerGroupService service =
getConsumerGroupService(new String[]{"--bootstrap-server",
clusterInstance.bootstrapServers(), "--list"})
) {
- Set<String> expectedGroups = set(List.of(topicPartitionsGroup,
topicGroup, protocolGroup));
- final AtomicReference<Set> foundGroups = new
AtomicReference<>();
+ Set<String> expectedGroups = Set.of(topicPartitionsGroup,
topicGroup, protocolGroup);
+ final AtomicReference<Set<String>> foundGroups = new
AtomicReference<>();
TestUtils.waitForCondition(() -> {
- foundGroups.set(set(service.listConsumerGroups()));
+ foundGroups.set(new
HashSet<>(service.listConsumerGroups()));
return Objects.equals(expectedGroups, foundGroups.get());
}, () -> "Expected --list to show groups " + expectedGroups +
", but found " + foundGroups.get() + ".");
}
- removeConsumer(set(List.of(topicPartitionsGroup, topicGroup,
protocolGroup)));
- deleteTopic(topic);
+ removeConsumer(clusterInstance, Set.of(topicPartitionsGroup,
topicGroup, protocolGroup));
+ deleteTopic(clusterInstance, topic);
}
}
- @ClusterTest
+ @Test
Review Comment:
Updated as suggested. This test has been moved to `ConsumerGroupCommandTest`.
--
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]
