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