This is an automated email from the ASF dual-hosted git repository.
bbejeck 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 9a520fd904c MINOR: reduce Kafka Streams integration test runtime
(#22833)
9a520fd904c is described below
commit 9a520fd904c0dd54c2606be38dfc0533075c4070
Author: Matthias J. Sax <[email protected]>
AuthorDate: Wed Jul 15 08:15:19 2026 -0700
MINOR: reduce Kafka Streams integration test runtime (#22833)
Some integrationt tests depend on session-timeout to expire. Instead of
reducing session timout to speed up the test, this PR change the test
code to send expliclit leave group requests on close().
Reviewers: Alieh Saeedi <[email protected]>, TengYao Chi
<[email protected]>, Bill Bejeck <[email protected]>
---
.../apache/kafka/streams/integration/IQv2IntegrationTest.java | 6 +++---
.../integration/KTableSourceTopicRestartIntegrationTest.java | 11 +++++------
.../streams/integration/PositionRestartIntegrationTest.java | 6 +++---
3 files changed, 11 insertions(+), 12 deletions(-)
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2IntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2IntegrationTest.java
index c2b532afc77..ebecbd12abb 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2IntegrationTest.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/IQv2IntegrationTest.java
@@ -27,6 +27,8 @@ import
org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.CloseOptions.GroupMembershipOperation;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
@@ -443,7 +445,7 @@ public class IQv2IntegrationTest {
// Discard the basic streams and replace with test-specific topology
if (kafkaStreams != null) {
- kafkaStreams.close();
+
kafkaStreams.close(CloseOptions.groupMembershipOperation(GroupMembershipOperation.LEAVE_GROUP));
}
final String safeTestName = safeUniqueTestName(testInfo);
this.groupProtocol = groupProtocol;
@@ -476,8 +478,6 @@ public class IQv2IntegrationTest {
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.Integer().getClass());
config.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
- config.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 200);
- config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
config.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, groupProtocol);
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KTableSourceTopicRestartIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KTableSourceTopicRestartIntegrationTest.java
index 52dd78ec17e..a5aabd1783e 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KTableSourceTopicRestartIntegrationTest.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/KTableSourceTopicRestartIntegrationTest.java
@@ -17,12 +17,13 @@
package org.apache.kafka.streams.integration;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.CloseOptions.GroupMembershipOperation;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
@@ -75,8 +76,6 @@ public class KTableSourceTopicRestartIntegrationTest {
STREAMS_CONFIG.put(StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG, 0);
STREAMS_CONFIG.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 5L);
STREAMS_CONFIG.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
WallclockTimestampExtractor.class);
- STREAMS_CONFIG.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
- STREAMS_CONFIG.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 300);
PRODUCER_CONFIG.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
PRODUCER_CONFIG.put(ProducerConfig.ACKS_CONFIG, "all");
@@ -128,7 +127,7 @@ public class KTableSourceTopicRestartIntegrationTest {
assertNumberValuesRead(readKeyValues, expectedInitialResultsMap,
"Table did not read all values");
- streams.close();
+
streams.close(CloseOptions.groupMembershipOperation(GroupMembershipOperation.LEAVE_GROUP));
streams = new KafkaStreams(streamsBuilder.build(), STREAMS_CONFIG);
// the state restore listener will append one record to the log
streams.setGlobalStateRestoreListener(new
UpdatingSourceTopicOnRestoreStartStateRestoreListener());
@@ -156,7 +155,7 @@ public class KTableSourceTopicRestartIntegrationTest {
assertNumberValuesRead(readKeyValues, expectedInitialResultsMap,
"Table did not read all values");
- streams.close();
+
streams.close(CloseOptions.groupMembershipOperation(GroupMembershipOperation.LEAVE_GROUP));
streams = new KafkaStreams(streamsBuilder.build(), STREAMS_CONFIG);
// the state restore listener will append one record to the log
streams.setGlobalStateRestoreListener(new
UpdatingSourceTopicOnRestoreStartStateRestoreListener());
@@ -182,7 +181,7 @@ public class KTableSourceTopicRestartIntegrationTest {
assertNumberValuesRead(readKeyValues, expectedInitialResultsMap,
"Table did not read all values");
- streams.close();
+
streams.close(CloseOptions.groupMembershipOperation(GroupMembershipOperation.LEAVE_GROUP));
streams = IntegrationTestUtils.getRunningStreams(STREAMS_CONFIG,
streamsBuilder, false);
produceKeyValues("f", "g", "h");
diff --git
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
index 588cdeba05e..612f9633256 100644
---
a/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
+++
b/streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/PositionRestartIntegrationTest.java
@@ -26,6 +26,8 @@ import
org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.Time;
+import org.apache.kafka.streams.CloseOptions;
+import org.apache.kafka.streams.CloseOptions.GroupMembershipOperation;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
@@ -385,7 +387,7 @@ public class PositionRestartIntegrationTest {
shouldReachExpectedPosition(query);
// reboot
- kafkaStreams.close();
+
kafkaStreams.close(CloseOptions.groupMembershipOperation(GroupMembershipOperation.LEAVE_GROUP));
kafkaStreams = IntegrationTestUtils.getStartedStreams(streamsConfig,
streamsBuilder, false);
shouldReachExpectedPosition(query);
@@ -666,8 +668,6 @@ public class PositionRestartIntegrationTest {
config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,
Serdes.Integer().getClass());
config.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
- config.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 200);
- config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 1000);
config.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 100L);
config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 1);
config.put(InternalConfig.IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED, true);
