You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2026/06/29 14:14:48 UTC
[PR] KAFKA-20553: Use ClusterInstance for streams group command tests [kafka]
FrankYang0529 opened a new pull request, #22701:
URL: https://github.com/apache/kafka/pull/22701
Followup from: https://github.com/apache/kafka/pull/22521#discussion_r3491032876
Migrate the streams group command integration tests from `EmbeddedKafkaCluster` to the `ClusterInstance` test framework:
- DescribeStreamsGroupTest
- ListStreamsGroupTest
- DeleteStreamsGroupTest
- DeleteStreamsGroupOffsetTest
- ResetStreamsGroupOffsetTest
--
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] KAFKA-20553: Use ClusterInstance for streams group command tests [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #22701:
URL: https://github.com/apache/kafka/pull/22701#discussion_r3520298814
##########
tools/src/test/java/org/apache/kafka/tools/streams/ListStreamsGroupTest.java:
##########
@@ -54,73 +52,60 @@
import joptsimple.OptionException;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+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.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
import static org.junit.jupiter.api.Assertions.assertEquals;
-@Timeout(600)
-@Tag("integration")
+@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 = "0"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "100"),
+ }
+)
public class ListStreamsGroupTest {
-
- public static EmbeddedKafkaCluster cluster = null;
- static KafkaStreams streams;
private static final String APP_ID = "streams-group-command-test";
private static final String INPUT_TOPIC = "customInputTopic";
private static final String OUTPUT_TOPIC = "customOutputTopic";
- @BeforeAll
- public static void setup() throws Exception {
- // start the cluster and create the input topic
- final Properties props = new Properties();
- cluster = new EmbeddedKafkaCluster(1, props);
- cluster.start();
- cluster.createTopic(INPUT_TOPIC, 2, 1);
+ private final ClusterInstance cluster;
-
- // start kafka streams
- Properties streamsProp = new Properties();
- streamsProp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
- streamsProp.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
- streamsProp.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
- streamsProp.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
- streamsProp.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
- streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
- streamsProp.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);
- streamsProp.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
-
- streams = new KafkaStreams(topology(), streamsProp);
- startApplicationAndWaitUntilRunning(streams);
+ ListStreamsGroupTest(ClusterInstance cluster) {
+ this.cluster = cluster;
}
- @AfterAll
- public static void closeCluster() {
- streams.close();
- cluster.stop();
- cluster = null;
- }
-
- @Test
+ @ClusterTest
public void testListStreamsGroupWithoutFilters() throws Exception {
- try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list"})) {
+ cluster.createTopic(INPUT_TOPIC, 2, (short) 1);
+ try (KafkaStreams ignored = startStreamsApp();
+ StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list"})) {
Set<String> expectedGroups = Set.of(APP_ID);
final AtomicReference<Set> foundGroups = new AtomicReference<>();
TestUtils.waitForCondition(() -> {
foundGroups.set(new HashSet<>(service.listStreamsGroups()));
return Objects.equals(expectedGroups, foundGroups.get());
}, () -> "Expected --list to show streams groups " + expectedGroups + ", but found " + foundGroups.get() + ".");
-
}
}
- @Test
+ @ClusterTest
Review Comment:
`testListStreamsGroupWithStates` is not argument check test case, I think we cannot just use `@Test`.
--
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] KAFKA-20553: Use ClusterInstance for streams group command tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22701:
URL: https://github.com/apache/kafka/pull/22701#discussion_r3492766043
##########
tools/src/test/java/org/apache/kafka/tools/streams/DeleteStreamsGroupOffsetTest.java:
##########
@@ -57,175 +47,158 @@
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.kafka.common.GroupState.EMPTY;
+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.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.coordinator.transaction.TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG;
+import static org.apache.kafka.coordinator.transaction.TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG;
+import static org.apache.kafka.coordinator.transaction.TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@Timeout(600)
-@Tag("integration")
+@ClusterTestDefaults(
+ types = {Type.CO_KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "0"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "100"),
+ }
+)
public class DeleteStreamsGroupOffsetTest {
private static final String TOPIC_PREFIX = "foo-";
private static final String APP_ID_PREFIX = "streams-group-command-test";
private static final int RECORD_TOTAL = 5;
- public static EmbeddedKafkaCluster cluster;
- private static String bootstrapServers;
private static final String OUTPUT_TOPIC_PREFIX = "output-topic-";
- @BeforeAll
- public static void startCluster() {
- final Properties props = new Properties();
- cluster = new EmbeddedKafkaCluster(2, props);
- cluster.start();
-
- bootstrapServers = cluster.bootstrapServers();
- }
-
- @AfterEach
- public void deleteTopicsAndGroups() {
- try (final Admin adminClient = cluster.createAdminClient()) {
- // delete all topics
- final Set<String> topics = adminClient.listTopics().names().get();
- adminClient.deleteTopics(topics).all().get();
- // delete all groups
- List<String> groupIds =
- adminClient.listGroups(ListGroupsOptions.forStreamsGroups().timeoutMs(1000)).all().get()
- .stream().map(GroupListing::groupId).toList();
- adminClient.deleteStreamsGroups(groupIds).all().get();
- } catch (final UnknownTopicOrPartitionException ignored) {
- } catch (final ExecutionException | InterruptedException e) {
- if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) {
- throw new RuntimeException(e);
- }
- }
- }
-
- private Properties createStreamsConfig(String bootstrapServers, String appId) {
+ private static Properties createStreamsConfig(String bootstrapServers, String appId) {
final Properties configs = new Properties();
configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
configs.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
configs.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
configs.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
- configs.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
+ configs.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");
Review Comment:
Why do we need this change?
--
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] KAFKA-20553: Use ClusterInstance for streams group command tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22701:
URL: https://github.com/apache/kafka/pull/22701#discussion_r3498368042
##########
tools/src/test/java/org/apache/kafka/tools/streams/ListStreamsGroupTest.java:
##########
@@ -54,73 +52,60 @@
import joptsimple.OptionException;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+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.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
import static org.junit.jupiter.api.Assertions.assertEquals;
-@Timeout(600)
-@Tag("integration")
+@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 = "0"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "100"),
+ }
+)
public class ListStreamsGroupTest {
-
- public static EmbeddedKafkaCluster cluster = null;
- static KafkaStreams streams;
private static final String APP_ID = "streams-group-command-test";
private static final String INPUT_TOPIC = "customInputTopic";
private static final String OUTPUT_TOPIC = "customOutputTopic";
- @BeforeAll
- public static void setup() throws Exception {
- // start the cluster and create the input topic
- final Properties props = new Properties();
- cluster = new EmbeddedKafkaCluster(1, props);
- cluster.start();
- cluster.createTopic(INPUT_TOPIC, 2, 1);
+ private final ClusterInstance cluster;
-
- // start kafka streams
- Properties streamsProp = new Properties();
- streamsProp.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
- streamsProp.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, cluster.bootstrapServers());
- streamsProp.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
- streamsProp.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
- streamsProp.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
- streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, APP_ID);
- streamsProp.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 2);
- streamsProp.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
-
- streams = new KafkaStreams(topology(), streamsProp);
- startApplicationAndWaitUntilRunning(streams);
+ ListStreamsGroupTest(ClusterInstance cluster) {
+ this.cluster = cluster;
}
- @AfterAll
- public static void closeCluster() {
- streams.close();
- cluster.stop();
- cluster = null;
- }
-
- @Test
+ @ClusterTest
public void testListStreamsGroupWithoutFilters() throws Exception {
- try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list"})) {
+ cluster.createTopic(INPUT_TOPIC, 2, (short) 1);
+ try (KafkaStreams ignored = startStreamsApp();
+ StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list"})) {
Set<String> expectedGroups = Set.of(APP_ID);
final AtomicReference<Set> foundGroups = new AtomicReference<>();
TestUtils.waitForCondition(() -> {
foundGroups.set(new HashSet<>(service.listStreamsGroups()));
return Objects.equals(expectedGroups, foundGroups.get());
}, () -> "Expected --list to show streams groups " + expectedGroups + ", but found " + foundGroups.get() + ".");
-
}
}
- @Test
+ @ClusterTest
Review Comment:
It does not need the `ClusterInstance`, right?
##########
tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTestUtils.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.tools.streams;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.KeyValueTimestamp;
+import org.apache.kafka.streams.StreamsBuilder;
+import org.apache.kafka.test.TestUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+public final class StreamsGroupCommandTestUtils {
+
+ private StreamsGroupCommandTestUtils() {
+ }
+
+ public static KafkaStreams getStartedStreams(final Properties streamsConfig, final StreamsBuilder builder, final boolean clean) {
+ final KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfig);
+ if (clean) {
+ streams.cleanUp();
+ }
+ streams.start();
+ return streams;
+ }
+
+ public static void startApplicationAndWaitUntilRunning(final KafkaStreams streams) throws InterruptedException {
+ streams.start();
+ TestUtils.waitForCondition(
+ () -> streams.state() == KafkaStreams.State.RUNNING,
+ "Streams application did not reach the RUNNING state.");
+ }
+
+ public static void produceSynchronously(final String bootstrapServers,
+ final String topic,
+ final Optional<Integer> partition,
+ final List<KeyValueTimestamp<String, String>> toProduce) {
+ Properties producerConfig = TestUtils.producerConfig(bootstrapServers, StringSerializer.class, StringSerializer.class);
+ try (Producer<String, String> producer = new KafkaProducer<>(producerConfig)) {
+ List<Future<RecordMetadata>> futures = new ArrayList<>();
+ for (KeyValueTimestamp<String, String> record : toProduce) {
+ futures.add(producer.send(
+ new ProducerRecord<>(topic, partition.orElse(null), record.timestamp(), record.key(), record.value(), null)));
Review Comment:
```java
public static void produce(final ClusterInstance instance,
final String topic,
final Optional<Integer> partition,
final List<KeyValueTimestamp<String, String>> toProduce) {
try (Producer<String, String> producer = instance.producer(Map.of(
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName(),
ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()
))) {
var futures = toProduce.stream()
.map(record -> producer.send(new ProducerRecord<>(topic, partition.orElse(null),
record.timestamp(), record.key(), record.value())))
.toList();
producer.flush();
futures.forEach(f -> Assertions.assertDoesNotThrow(() -> f.get()));
}
}
```
##########
tools/src/test/java/org/apache/kafka/tools/streams/ListStreamsGroupTest.java:
##########
@@ -138,63 +123,75 @@ public void testListStreamsGroupWithStates() throws Exception {
}
}
- @Test
+ @ClusterTest
public void testListStreamsGroupWithSpecifiedStates() throws Exception {
- try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list", "--state", "stable"})) {
- Set<GroupListing> expectedListing = Set.of(
- new GroupListing(
- APP_ID,
- Optional.of(GroupType.STREAMS),
- "streams",
- Optional.of(GroupState.STABLE))
- );
-
- final AtomicReference<Set<GroupListing>> foundListing = new AtomicReference<>();
-
- TestUtils.waitForCondition(() -> {
- foundListing.set(new HashSet<>(service.listStreamsGroupsInStates(Set.of())));
- return Objects.equals(expectedListing, foundListing.get());
- }, () -> "Expected --list to show streams groups " + expectedListing + ", but found " + foundListing.get() + ".");
- }
+ cluster.createTopic(INPUT_TOPIC, 2, (short) 1);
+ try (KafkaStreams ignored = startStreamsApp()) {
+ try (StreamsGroupCommand.StreamsGroupService service = getStreamsGroupService(new String[]{"--bootstrap-server", cluster.bootstrapServers(), "--list", "--state", "stable"})) {
Review Comment:
Could we consolidate these try-catch blocks into a single one?
##########
tools/src/test/java/org/apache/kafka/tools/streams/DescribeStreamsGroupTest.java:
##########
@@ -51,74 +48,64 @@
import joptsimple.OptionException;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+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.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@Timeout(600)
-@Tag("integration")
+@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 = "0"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "100"),
+ }
+)
public class DescribeStreamsGroupTest {
- public static EmbeddedKafkaCluster cluster = null;
- static KafkaStreams streams;
private static final String APP_ID = "streams-group-command-test";
private static final String APP_ID_2 = "streams-group-command-test-2";
private static final String INPUT_TOPIC = "customInputTopic";
private static final String OUTPUT_TOPIC = "customOutputTopic";
private static final String INPUT_TOPIC_2 = "customInputTopic2";
private static final String OUTPUT_TOPIC_2 = "customOutputTopic2";
- private static String bootstrapServers;
-
- @BeforeAll
- public static void setup() throws Exception {
- // start the cluster and create the input topic
- final Properties props = new Properties();
- cluster = new EmbeddedKafkaCluster(1, props);
- cluster.start();
- cluster.createTopic(INPUT_TOPIC, 2, 1);
- bootstrapServers = cluster.bootstrapServers();
-
-
- // start kafka streams
- Properties streamsProp = streamsProp(APP_ID);
- streams = new KafkaStreams(topology(INPUT_TOPIC, OUTPUT_TOPIC), streamsProp);
- startApplicationAndWaitUntilRunning(streams);
- }
- @AfterAll
- public static void closeCluster() {
- streams.close();
- cluster.deleteTopics(INPUT_TOPIC, OUTPUT_TOPIC, INPUT_TOPIC_2, OUTPUT_TOPIC_2);
- cluster.stop();
- cluster = null;
+ private final ClusterInstance cluster;
+
+ DescribeStreamsGroupTest(ClusterInstance cluster) {
+ this.cluster = cluster;
}
- @Test
+ @ClusterTest
Review Comment:
ditto. Does it need to have a cluster instance?
--
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] KAFKA-20553: Use ClusterInstance for streams group command tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on PR #22701:
URL: https://github.com/apache/kafka/pull/22701#issuecomment-5017052320
The execution time is vastly different for `DescribeStreamsGroupTest`
*before*
```
Child Tests Failures Skipped Duration Success rate
[testDescribeMultipleStreamsGroupWithMembersAndVerboseOptions()](https://github.com/apache/kafka/pull/BOf1cqoJITM.html) 1 0 0 5.862s 100%
[testDescribeNonExistingStreamsGroup()](https://github.com/apache/kafka/pull/BAYvWcALVQQ.html) 1 0 0 0.005s 100%
testDescribeStreamsGroup() 1 0 0 0.011s 100%
testDescribeStreamsGroupWithMembersAndVerboseOptions() 1 0 0 0.006s 100%
testDescribeStreamsGroupWithMembersOption() 1 0 0 0.004s 100%
testDescribeStreamsGroupWithStateAndVerboseOptions() 1 0 0 0.007s 100%
testDescribeStreamsGroupWithStateOption() 1 0 0 0.003s 100%
testDescribeStreamsGroupWithTopologyOption() 1 0 0 0.005s 100%
testDescribeStreamsGroupWithVerboseOption() 1 0 0 0.013s 100%
[testDescribeWithoutGroupOption()](https://github.com/apache/kafka/pull/AFNfcCNTpLg.html) 1 0 0 0.001s 100%
testDescribeWithUnrecognizedOption() 1 0 0 0.001s 100%
```
*after*
```
[testDescribeMultipleStreamsGroupWithMembersAndVerboseOptions(ClusterInstance)](https://github.com/apache/kafka/pull/Wope8t1GL2E/index.html) 1 0 0 12.313s 100%
[testDescribeNonExistingStreamsGroup(ClusterInstance)](https://github.com/apache/kafka/pull/wbc0Lyi1wFQ/index.html) 1 0 0 0.484s 100%
[testDescribeStreamsGroup(ClusterInstance)](https://github.com/apache/kafka/pull/I9ZyO4tnZAw/index.html) 1 0 0 8.347s 100%
[testDescribeStreamsGroupWithMembersAndVerboseOptions(ClusterInstance)](https://github.com/apache/kafka/pull/7bf_RQXomvE/index.html) 1 0 0 6.278s 100%
[testDescribeStreamsGroupWithMembersOption(ClusterInstance)](https://github.com/apache/kafka/pull/GyGJhWCoyNo/index.html) 1 0 0 6.259s 100%
[testDescribeStreamsGroupWithStateAndVerboseOptions(ClusterInstance)](https://github.com/apache/kafka/pull/DX1r17KtEXQ/index.html) 1 0 0 6.271s 100%
[testDescribeStreamsGroupWithStateOption(ClusterInstance)](https://github.com/apache/kafka/pull/5pXzE59AlOU/index.html) 1 0 0 6.271s 100%
[testDescribeStreamsGroupWithTopologyOption(ClusterInstance)](https://github.com/apache/kafka/pull/K4QL2pFOp_8/index.html) 1 0 0 6.398s 100%
[testDescribeStreamsGroupWithVerboseOption(ClusterInstance)](https://github.com/apache/kafka/pull/l3Fz5Qc2NA0/index.html) 1 0 0 6.652s 100%
[testDescribeWithoutGroupOption()](https://github.com/apache/kafka/pull/AFNfcCNTpLg.html) 1 0 0 0.013s 100%
testDescribeWithUnrecognizedOption() 1 0 0 0.002s 100%
```
It would be a good follow-up to combine some tests, like `testDescribeStreamsGroup` and `testDescribeStreamsGroupWithVerboseOption`.
--
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] KAFKA-20553: Use ClusterInstance for streams group command tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22701:
URL: https://github.com/apache/kafka/pull/22701#discussion_r3611190650
##########
tools/src/test/java/org/apache/kafka/tools/streams/DeleteStreamsGroupOffsetTest.java:
##########
@@ -57,78 +51,65 @@
import java.util.Objects;
import java.util.Optional;
import java.util.Properties;
-import java.util.Set;
-import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.kafka.common.GroupState.EMPTY;
+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.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG;
+import static org.apache.kafka.coordinator.transaction.TransactionLogConfig.TRANSACTIONS_TOPIC_MIN_ISR_CONFIG;
+import static org.apache.kafka.coordinator.transaction.TransactionLogConfig.TRANSACTIONS_TOPIC_PARTITIONS_CONFIG;
+import static org.apache.kafka.coordinator.transaction.TransactionLogConfig.TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
-@Timeout(600)
-@Tag("integration")
+@ClusterTestDefaults(
+ types = {Type.CO_KRAFT},
+ brokers = 2,
+ serverProperties = {
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = OFFSETS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TRANSACTIONS_TOPIC_PARTITIONS_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TRANSACTIONS_TOPIC_REPLICATION_FACTOR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = TRANSACTIONS_TOPIC_MIN_ISR_CONFIG, value = "1"),
+ @ClusterConfigProperty(key = GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG, value = "0"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "100"),
+ @ClusterConfigProperty(key = STREAMS_GROUP_MIN_HEARTBEAT_INTERVAL_MS_CONFIG, value = "100"),
+ }
+)
public class DeleteStreamsGroupOffsetTest {
private static final String TOPIC_PREFIX = "foo-";
private static final String APP_ID_PREFIX = "streams-group-command-test";
private static final int RECORD_TOTAL = 5;
- public static EmbeddedKafkaCluster cluster;
- private static String bootstrapServers;
private static final String OUTPUT_TOPIC_PREFIX = "output-topic-";
- @BeforeAll
- public static void startCluster() {
- final Properties props = new Properties();
- cluster = new EmbeddedKafkaCluster(2, props);
- cluster.start();
-
- bootstrapServers = cluster.bootstrapServers();
- }
-
- @AfterEach
- public void deleteTopicsAndGroups() {
- try (final Admin adminClient = cluster.createAdminClient()) {
- // delete all topics
- final Set<String> topics = adminClient.listTopics().names().get();
- adminClient.deleteTopics(topics).all().get();
- // delete all groups
- List<String> groupIds =
- adminClient.listGroups(ListGroupsOptions.forStreamsGroups().timeoutMs(1000)).all().get()
- .stream().map(GroupListing::groupId).toList();
- adminClient.deleteStreamsGroups(groupIds).all().get();
- } catch (final UnknownTopicOrPartitionException ignored) {
- } catch (final ExecutionException | InterruptedException e) {
- if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) {
- throw new RuntimeException(e);
- }
- }
- }
-
- private Properties createStreamsConfig(String bootstrapServers, String appId) {
+ private static Properties createStreamsConfig(String bootstrapServers, String appId) {
final Properties configs = new Properties();
configs.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
configs.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
configs.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
configs.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
configs.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, GroupProtocol.STREAMS.name().toLowerCase(Locale.getDefault()));
configs.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
+ // Use a unique state directory per instance. TestUtils.randomString uses a fixed seed, so app ids are
Review Comment:
spot on!
--
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] KAFKA-20553: Use ClusterInstance for streams group command tests [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #22701:
URL: https://github.com/apache/kafka/pull/22701
--
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