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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
