You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "m1a2st (via GitHub)" <gi...@apache.org> on 2026/07/19 09:25:50 UTC
[PR] KAFKA-20553: Use ClusterInstance for streams AbstractResetIntegrationTest [kafka]
m1a2st opened a new pull request, #22871:
URL: https://github.com/apache/kafka/pull/22871
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:
- ResetIntegrationTest.java
- AbstractResetIntegrationTest.java
- ResetIntegrationWithSslTest.java
--
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 ResetIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22871:
URL: https://github.com/apache/kafka/pull/22871#discussion_r3648766383
##########
tools/src/test/java/org/apache/kafka/tools/ResetIntegrationTest.java:
##########
@@ -45,91 +61,164 @@
import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.util.ArrayList;
import java.util.Calendar;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
-
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.isEmptyConsumerGroup;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForEmptyConsumerGroup;
-import static org.apache.kafka.streams.utils.TestUtils.safeUniqueTestName;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+
+import static java.time.Duration.ofMillis;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_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.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Tests local state store and global application cleanup.
*/
-@Timeout(600)
-public class ResetIntegrationTest extends AbstractResetIntegrationTest {
+@ClusterTestDefaults(
+ types = {Type.CO_KRAFT},
+ brokers = 3,
+ 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 = GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "0"),
+ }
+)
+public class ResetIntegrationTest {
private static final String NON_EXISTING_TOPIC = "nonExistingTopic";
- public static final EmbeddedKafkaCluster CLUSTER;
+ private static final String INPUT_TOPIC = "inputTopic";
+ private static final String OUTPUT_TOPIC = "outputTopic";
+ private static final String OUTPUT_TOPIC_2 = "outputTopic2";
+ private static final String OUTPUT_TOPIC_2_RERUN = "outputTopic2_rerun";
+ private static final String INTERMEDIATE_USER_TOPIC = "userTopic";
+
+ private static final int STREAMS_CONSUMER_TIMEOUT = 2000;
+ private static final int CLEANUP_CONSUMER_TIMEOUT = 2000;
+ private static final long DEFAULT_TIMEOUT_MS = 60_000L;
+ private static final String APP_ID_PREFIX = "reset-integration-test";
+
+ private Properties streamsConfig;
+ private Map<String, Object> resultConsumerConfig;
+ private Map<String, Object> producerConfig;
+ private long recordTimestamp;
+
+ @ClusterTest
+ public void testResetWhenInternalTopicsAreSpecified(ClusterInstance cluster) throws Exception {
+ try (Admin admin = cluster.admin()) {
+ final String appId = generateAppId();
+ prepare(cluster, null, appId);
+ runResetWhenInternalTopicsAreSpecified(cluster, admin, null, appId);
+ }
+ }
- static {
- final Properties brokerProps = new Properties();
- // we double the value passed to `time.sleep` in each iteration in one of the map functions, so we disable
- // expiration of connections by the brokers to avoid errors when `AdminClient` sends requests after potentially
- // very long sleep times
- brokerProps.put(SocketServerConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG, -1L);
- CLUSTER = new EmbeddedKafkaCluster(3, brokerProps);
+ @ClusterTest
+ public void testReprocessingFromScratchAfterResetWithoutIntermediateUserTopic(ClusterInstance cluster) throws Exception {
+ try (Admin admin = cluster.admin()) {
+ final String appId = generateAppId();
+ prepare(cluster, null, appId);
+ runReprocessingFromScratchWithoutIntermediateUserTopic(cluster, admin, null, appId);
+ }
}
- @BeforeAll
- public static void startCluster() throws IOException {
- CLUSTER.start();
+ @ClusterTest
+ public void testReprocessingFromScratchAfterResetWithIntermediateUserTopic(ClusterInstance cluster) throws Exception {
+ try (Admin admin = cluster.admin()) {
+ final String appId = generateAppId();
+ prepare(cluster, null, appId);
+ runReprocessingFromScratchWithIntermediateUserTopic(cluster, admin, null, false, appId);
+ }
}
- @AfterAll
- public static void closeCluster() {
- CLUSTER.stop();
+ @ClusterTest
+ public void testReprocessingFromScratchAfterResetWithIntermediateInternalTopic(ClusterInstance cluster) throws Exception {
+ try (Admin admin = cluster.admin()) {
+ final String appId = generateAppId();
+ prepare(cluster, null, appId);
+ runReprocessingFromScratchWithIntermediateUserTopic(cluster, admin, null, true, appId);
+ }
}
- @Override
- Map<String, Object> getClientSslConfig() {
- return null;
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SSL, controllerSecurityProtocol = SecurityProtocol.SSL)
+ public void testResetWhenInternalTopicsAreSpecifiedWithSsl(ClusterInstance cluster) throws Exception {
+ final Map<String, Object> sslConfig = cluster.setClientSslConfig(new HashMap<>());
+ try (Admin admin = cluster.admin()) {
+ final String appId = generateAppId();
+ prepare(cluster, sslConfig, appId);
+ runResetWhenInternalTopicsAreSpecified(cluster, admin, sslConfig, appId);
+ }
}
- @BeforeEach
- public void before(final TestInfo testInfo) throws Exception {
- cluster = CLUSTER;
- prepareTest(testInfo);
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SSL, controllerSecurityProtocol = SecurityProtocol.SSL)
+ public void testReprocessingFromScratchAfterResetWithoutIntermediateUserTopicWithSsl(ClusterInstance cluster) throws Exception {
+ final Map<String, Object> sslConfig = cluster.setClientSslConfig(new HashMap<>());
+ try (Admin admin = cluster.admin()) {
+ final String appId = generateAppId();
+ prepare(cluster, sslConfig, appId);
+ runReprocessingFromScratchWithoutIntermediateUserTopic(cluster, admin, sslConfig, appId);
+ }
}
- @AfterEach
- public void after() throws Exception {
- cleanupTest();
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SSL, controllerSecurityProtocol = SecurityProtocol.SSL)
+ public void testReprocessingFromScratchAfterResetWithIntermediateUserTopicWithSsl(ClusterInstance cluster) throws Exception {
+ final Map<String, Object> sslConfig = cluster.setClientSslConfig(new HashMap<>());
+ try (Admin admin = cluster.admin()) {
+ final String appId = generateAppId();
+ prepare(cluster, sslConfig, appId);
+ runReprocessingFromScratchWithIntermediateUserTopic(cluster, admin, sslConfig, false, appId);
+ }
}
- @Test
- public void shouldNotAllowToResetWhileStreamsIsRunning(final TestInfo testInfo) throws Exception {
- final String appID = safeUniqueTestName(testInfo);
+ @ClusterTest(brokerSecurityProtocol = SecurityProtocol.SSL, controllerSecurityProtocol = SecurityProtocol.SSL)
+ public void testReprocessingFromScratchAfterResetWithIntermediateInternalTopicWithSsl(ClusterInstance cluster) throws Exception {
Review Comment:
How about using `ClusterTests` ?
```java
@ClusterTests({
@ClusterTest,
@ClusterTest(brokerSecurityProtocol = SecurityProtocol.SSL, controllerSecurityProtocol = SecurityProtocol.SSL)
})
```
--
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 AbstractResetIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #22871:
URL: https://github.com/apache/kafka/pull/22871#discussion_r3646470214
##########
tools/src/test/java/org/apache/kafka/tools/ResetIntegrationWithSslTest.java:
##########
@@ -16,73 +16,85 @@
*/
package org.apache.kafka.tools;
-import org.apache.kafka.common.network.ConnectionMode;
-import org.apache.kafka.network.SocketServerConfigs;
-import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
-import org.apache.kafka.test.TestSslUtils;
-import org.apache.kafka.test.TestUtils;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.apache.kafka.common.test.ClusterInstance;
+import org.apache.kafka.common.test.api.ClusterConfigProperty;
+import org.apache.kafka.common.test.api.ClusterTest;
+import org.apache.kafka.common.test.api.ClusterTestDefaults;
+import org.apache.kafka.common.test.api.Type;
-import org.junit.jupiter.api.AfterAll;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.TestInfo;
-
-import java.io.IOException;
+import java.util.HashMap;
import java.util.Map;
-import java.util.Properties;
+
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_INITIAL_REBALANCE_DELAY_MS_CONFIG;
+import static org.apache.kafka.coordinator.group.GroupCoordinatorConfig.GROUP_MIN_SESSION_TIMEOUT_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;
/**
* Tests command line SSL setup for reset tool.
*/
+@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 = GROUP_MIN_SESSION_TIMEOUT_MS_CONFIG, value = "0"),
+ }
+)
public class ResetIntegrationWithSslTest extends AbstractResetIntegrationTest {
Review Comment:
Since we are using `ClusterTest` now, it would be simpler to merge `ResetIntegrationWithSslTest` into `ResetIntegrationTest`
--
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 ResetIntegrationTest [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #22871:
URL: https://github.com/apache/kafka/pull/22871
--
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