You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by jy...@apache.org on 2026/08/26 06:15:31 UTC
(kafka) branch trunk updated: KAFKA-16245: Improve reliability of short initialization timeout tests (#23230)
This is an automated email from the ASF dual-hosted git repository.
m1a2st 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 191a7fb15f7 KAFKA-16245: Improve reliability of short initialization timeout tests (#23230)
191a7fb15f7 is described below
commit 191a7fb15f706f7eb77657008b0fc0512bb46d6e
Author: Andrew Schofield <as...@confluent.io>
AuthorDate: Wed Aug 26 07:15:23 2026 +0100
KAFKA-16245: Improve reliability of short initialization timeout tests (#23230)
The `DescribeConsumerGroupTest.test...WithShortInitializationTimeout`
tests are flaky. They have an inherent race condition, which generally
works. However, they rely on admin client calls timing out before the
consumer offsets topic has auto-created. Prior to this PR, the test
cases internally iterated over the CONSUMER and CLASSIC group types, and
tried to get both group protocols to fail with a timeout before the
topic had initialized. Occasionally this race was lost and the test
failed.
I considered using a fresh cluster for each group type and running the
tests twice. However, it seems that there's little benefit testing two
group types, so I just changed the tests to use the CONSUMER group type
only.
Reviewers: Ken Huang <s7...@gmail.com>
---
.../consumer/group/DescribeConsumerGroupTest.java | 160 ++++++++++-----------
1 file changed, 76 insertions(+), 84 deletions(-)
diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
index d65c494dfb5..652c2cc211e 100644
--- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
+++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/DescribeConsumerGroupTest.java
@@ -903,115 +903,107 @@ public class DescribeConsumerGroupTest {
@ClusterTest
public void testDescribeSimpleConsumerGroup(ClusterInstance clusterInstance) throws Exception {
this.clusterInstance = clusterInstance;
- // Ensure that the offsets of consumers which don't use group management are still displayed
- for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
- String topic = TOPIC_PREFIX + groupProtocol.name();
- String group = GROUP_PREFIX + groupProtocol.name();
- createTopic(topic, 2);
+ GroupProtocol groupProtocol = GroupProtocol.CLASSIC;
+ String topic = TOPIC_PREFIX + groupProtocol.name();
+ String group = GROUP_PREFIX + groupProtocol.name();
+ createTopic(topic, 2);
- try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(GroupProtocol.CLASSIC, group, Set.of(new TopicPartition(topic, 0), new TopicPartition(topic, 1)), Map.of());
- ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
- ) {
- TestUtils.waitForCondition(() -> {
- Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
- return res.getKey().map(s -> s.equals(GroupState.EMPTY)).orElse(false)
- && res.getValue().isPresent() && res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2;
- }, "Expected a stable group with two members in describe group state result.");
- }
+ try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, Set.of(new TopicPartition(topic, 0), new TopicPartition(topic, 1)), Map.of());
+ ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group})
+ ) {
+ TestUtils.waitForCondition(() -> {
+ Entry<Optional<GroupState>, Optional<Collection<PartitionAssignmentState>>> res = service.collectGroupOffsets(group);
+ return res.getKey().map(s -> s.equals(GroupState.EMPTY)).orElse(false)
+ && res.getValue().isPresent() && res.getValue().get().stream().filter(s -> Objects.equals(s.group(), group)).count() == 2;
+ }, "Expected a stable group with two members in describe group state result.");
}
}
@ClusterTest
public void testDescribeGroupWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
this.clusterInstance = clusterInstance;
- for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
- String topic = TOPIC_PREFIX + groupProtocol.name();
- createTopic(topic);
-
- // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
- // complete before the timeout expires
- List<String> describeType = DESCRIBE_TYPES.get(RANDOM.nextInt(DESCRIBE_TYPES.size()));
- String group = GROUP_PREFIX + groupProtocol.name() + String.join("", describeType);
-
- // set the group initialization timeout too low for the group to stabilize
- List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--timeout", "1", "--group", group));
- cgcArgs.addAll(describeType);
-
- // run one consumer in the group consuming from a single-partition topic
- try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
- ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))
- ) {
- ExecutionException e = assertThrows(ExecutionException.class, service::describeGroups);
- assertInstanceOf(TimeoutException.class, e.getCause());
- }
+ GroupProtocol groupProtocol = GroupProtocol.CONSUMER;
+ String topic = TOPIC_PREFIX + groupProtocol.name();
+ createTopic(topic);
+
+ // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
+ // complete before the timeout expires
+ List<String> describeType = DESCRIBE_TYPES.get(RANDOM.nextInt(DESCRIBE_TYPES.size()));
+ String group = GROUP_PREFIX + groupProtocol.name() + String.join("", describeType);
+
+ // set the group service timeout too low for the group to stabilize
+ List<String> cgcArgs = new ArrayList<>(List.of("--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"));
+ cgcArgs.addAll(describeType);
+
+ // run one consumer in the group consuming from a single-partition topic
+ try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
+ ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(cgcArgs.toArray(new String[0]))
+ ) {
+ ExecutionException e = assertThrows(ExecutionException.class, service::describeGroups);
+ assertInstanceOf(TimeoutException.class, e.getCause());
}
}
@ClusterTest
public void testDescribeGroupOffsetsWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
this.clusterInstance = clusterInstance;
- for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
- String topic = TOPIC_PREFIX + groupProtocol.name();
- String group = GROUP_PREFIX + groupProtocol.name();
- createTopic(topic);
-
- // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
- // complete before the timeout expires
-
- // run one consumer in the group consuming from a single-partition topic
- try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
- // set the group initialization timeout too low for the group to stabilize
- ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
- ) {
- Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupOffsets(group));
- assertEquals(TimeoutException.class, e.getCause().getClass());
- }
+ GroupProtocol groupProtocol = GroupProtocol.CONSUMER;
+ String topic = TOPIC_PREFIX + groupProtocol.name();
+ String group = GROUP_PREFIX + groupProtocol.name();
+ createTopic(topic);
+
+ // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
+ // complete before the timeout expires
+
+ // run one consumer in the group consuming from a single-partition topic
+ try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
+ // set the group service timeout too low for the group to stabilize
+ ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
+ ) {
+ Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupOffsets(group));
+ assertEquals(TimeoutException.class, e.getCause().getClass());
}
}
@ClusterTest
public void testDescribeGroupMembersWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
this.clusterInstance = clusterInstance;
- for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
- String topic = TOPIC_PREFIX + groupProtocol.name();
- String group = GROUP_PREFIX + groupProtocol.name();
- createTopic(topic);
-
- // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
- // complete before the timeout expires
-
- // run one consumer in the group consuming from a single-partition topic
- try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
- // set the group initialization timeout too low for the group to stabilize
- ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
- ) {
- Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupMembers(group));
- assertEquals(TimeoutException.class, e.getCause().getClass());
- e = assertThrows(ExecutionException.class, () -> service.collectGroupMembers(group));
- assertEquals(TimeoutException.class, e.getCause().getClass());
- }
+ GroupProtocol groupProtocol = GroupProtocol.CONSUMER;
+ String topic = TOPIC_PREFIX + groupProtocol.name();
+ String group = GROUP_PREFIX + groupProtocol.name();
+ createTopic(topic);
+
+ // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
+ // complete before the timeout expires
+
+ // run one consumer in the group consuming from a single-partition topic
+ try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
+ // set the group service timeout too low for the group to stabilize
+ ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
+ ) {
+ Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupMembers(group));
+ assertEquals(TimeoutException.class, e.getCause().getClass());
}
}
@ClusterTest
public void testDescribeGroupStateWithShortInitializationTimeout(ClusterInstance clusterInstance) throws Exception {
this.clusterInstance = clusterInstance;
- for (GroupProtocol groupProtocol: clusterInstance.supportedGroupProtocols()) {
- String topic = TOPIC_PREFIX + groupProtocol.name();
- String group = GROUP_PREFIX + groupProtocol.name();
- createTopic(topic);
-
- // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
- // complete before the timeout expires
-
- // run one consumer in the group consuming from a single-partition topic
- try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
- // set the group initialization timeout too low for the group to stabilize
- ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
- ) {
- Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupState(group));
- assertEquals(TimeoutException.class, e.getCause().getClass());
- }
+ GroupProtocol groupProtocol = GroupProtocol.CONSUMER;
+ String topic = TOPIC_PREFIX + groupProtocol.name();
+ String group = GROUP_PREFIX + groupProtocol.name();
+ createTopic(topic);
+
+ // Let creation of the offsets topic happen during group initialization to ensure that initialization doesn't
+ // complete before the timeout expires
+
+ // run one consumer in the group consuming from a single-partition topic
+ try (AutoCloseable protocolConsumerGroupExecutor = consumerGroupClosable(groupProtocol, group, topic, Map.of());
+ // set the group service timeout too low for the group to stabilize
+ ConsumerGroupCommand.ConsumerGroupService service = consumerGroupService(new String[]{"--bootstrap-server", clusterInstance.bootstrapServers(), "--describe", "--group", group, "--timeout", "1"})
+ ) {
+ Throwable e = assertThrows(ExecutionException.class, () -> service.collectGroupState(group));
+ assertEquals(TimeoutException.class, e.getCause().getClass());
}
}