You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ab...@apache.org on 2023/01/22 23:58:23 UTC
[kafka] branch trunk updated: MINOR: fix flaky integrations tests by using 60s default timeout for startup (#13141)
This is an automated email from the ASF dual-hosted git repository.
ableegoldman 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 0601fa0935b MINOR: fix flaky integrations tests by using 60s default timeout for startup (#13141)
0601fa0935b is described below
commit 0601fa0935b03318ed08bfc578685fc4155ee948
Author: A. Sophie Blee-Goldman <so...@confluent.io>
AuthorDate: Sun Jan 22 15:57:58 2023 -0800
MINOR: fix flaky integrations tests by using 60s default timeout for startup (#13141)
The timeouts used for starting up Streams and waiting for the RUNNING state are all over the place across our integration tests, with some as low as 15s (which are unsurprisingly rather flaky). We use 60s as the default timeout for other APIs in the IntegrationTestUtils so we should do the same for #startApplicationAndWaitUntilRunning
I also noticed that we have several versions of that exact API in StreamsTestUtils, so I migrated everyone over to the IntegrationTestUtils#startApplicationAndWaitUntilRunning and added a few overloads for ease of use, including one for single KafkaStreams apps and one for using the default timeout
Reviewers: Matthias J. Sax <mj...@apache.org>
---
.../ConsistencyVectorIntegrationTest.java | 32 ++++----
.../integration/EmitOnChangeIntegrationTest.java | 3 +-
.../streams/integration/EosIntegrationTest.java | 24 +++---
.../integration/InternalTopicIntegrationTest.java | 59 +++++++-------
...yInnerJoinCustomPartitionerIntegrationTest.java | 23 +++---
.../KafkaStreamsCloseOptionsIntegrationTest.java | 3 +-
.../integration/LagFetchIntegrationTest.java | 3 +-
.../integration/NamedTopologyIntegrationTest.java | 81 +++++++++---------
.../OptimizedKTableIntegrationTest.java | 95 ++++++++++++----------
.../integration/QueryableStateIntegrationTest.java | 32 ++++----
.../integration/RangeQueryIntegrationTest.java | 4 +-
.../integration/RegexSourceIntegrationTest.java | 3 +-
.../integration/RestoreIntegrationTest.java | 61 ++++++++------
.../integration/RocksDBMetricsIntegrationTest.java | 5 +-
.../integration/StandbyTaskEOSIntegrationTest.java | 9 +-
.../integration/StoreQueryIntegrationTest.java | 10 +--
...amsUncaughtExceptionHandlerIntegrationTest.java | 47 ++++++-----
.../integration/TaskMetadataIntegrationTest.java | 6 +-
.../integration/utils/IntegrationTestUtils.java | 14 +++-
.../internals/DefaultStateUpdaterTest.java | 6 +-
.../org/apache/kafka/test/StreamsTestUtils.java | 27 +-----
21 files changed, 271 insertions(+), 276 deletions(-)
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/ConsistencyVectorIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/ConsistencyVectorIntegrationTest.java
index 78629cc652a..977f325fc40 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/ConsistencyVectorIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/ConsistencyVectorIntegrationTest.java
@@ -52,7 +52,6 @@ import org.junit.rules.TestName;
import org.junit.rules.Timeout;
import java.io.IOException;
-import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -121,24 +120,29 @@ public class ConsistencyVectorIntegrationTest {
final KafkaStreams kafkaStreams2 = createKafkaStreams(builder, streamsConfiguration());
final List<KafkaStreams> kafkaStreamsList = Arrays.asList(kafkaStreams1, kafkaStreams2);
- startApplicationAndWaitUntilRunning(kafkaStreamsList, Duration.ofSeconds(60));
+ try {
+ startApplicationAndWaitUntilRunning(kafkaStreamsList);
- produceValueRange(key, 0, batch1NumMessages);
+ produceValueRange(key, 0, batch1NumMessages);
- // Assert that all messages in the first batch were processed in a timely manner
- assertThat(semaphore.tryAcquire(batch1NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));
+ // Assert that all messages in the first batch were processed in a timely manner
+ assertThat(semaphore.tryAcquire(batch1NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));
- final QueryableStoreType<ReadOnlyKeyValueStore<Integer, Integer>> queryableStoreType = keyValueStore();
+ final QueryableStoreType<ReadOnlyKeyValueStore<Integer, Integer>> queryableStoreType = keyValueStore();
- // Assert that both active and standby have the same position bound
- final StateQueryRequest<ValueAndTimestamp<Integer>> request =
- StateQueryRequest
- .inStore(TABLE_NAME)
- .withQuery(KeyQuery.<Integer, ValueAndTimestamp<Integer>>withKey(key))
- .withPositionBound(PositionBound.unbounded());
+ // Assert that both active and standby have the same position bound
+ final StateQueryRequest<ValueAndTimestamp<Integer>> request =
+ StateQueryRequest
+ .inStore(TABLE_NAME)
+ .withQuery(KeyQuery.<Integer, ValueAndTimestamp<Integer>>withKey(key))
+ .withPositionBound(PositionBound.unbounded());
- checkPosition(batch1NumMessages, request, kafkaStreams1);
- checkPosition(batch1NumMessages, request, kafkaStreams2);
+ checkPosition(batch1NumMessages, request, kafkaStreams1);
+ checkPosition(batch1NumMessages, request, kafkaStreams2);
+ } finally {
+ kafkaStreams1.close();
+ kafkaStreams2.close();
+ }
}
private void checkPosition(final int batch1NumMessages,
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EmitOnChangeIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EmitOnChangeIntegrationTest.java
index 6945d708b87..876c5ed0190 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/EmitOnChangeIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/EmitOnChangeIntegrationTest.java
@@ -30,7 +30,6 @@ import org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThr
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.streams.integration.utils.IntegrationTestUtils;
import org.apache.kafka.streams.kstream.Materialized;
-import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
@@ -114,7 +113,7 @@ public class EmitOnChangeIntegrationTest {
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
kafkaStreams.setUncaughtExceptionHandler(exception -> StreamThreadExceptionResponse.REPLACE_THREAD);
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
inputTopic,
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
index 7fd07b01ff6..3ebcbdebef5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/EosIntegrationTest.java
@@ -87,9 +87,9 @@ import java.util.concurrent.atomic.AtomicReference;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForEmptyConsumerGroup;
import static org.apache.kafka.streams.query.StateQueryRequest.inStore;
-import static org.apache.kafka.test.StreamsTestUtils.startKafkaStreamsAndWaitForRunningState;
import static org.apache.kafka.test.TestUtils.consumerConfig;
import static org.apache.kafka.test.TestUtils.waitForCondition;
import static org.hamcrest.CoreMatchers.equalTo;
@@ -281,7 +281,7 @@ public class EosIntegrationTest {
);
try (final KafkaStreams streams = new KafkaStreams(builder.build(), config)) {
- startKafkaStreamsAndWaitForRunningState(streams, MAX_WAIT_TIME_MS);
+ startApplicationAndWaitUntilRunning(streams);
final List<KeyValue<Long, Long>> committedRecords = readResult(outputTopic, inputData.size(), CONSUMER_GROUP_ID);
checkResultPerKey(committedRecords, inputData, "The committed records do not match what expected");
@@ -339,7 +339,7 @@ public class EosIntegrationTest {
properties);
try (final KafkaStreams streams = new KafkaStreams(builder.build(), config)) {
- startKafkaStreamsAndWaitForRunningState(streams, MAX_WAIT_TIME_MS);
+ startApplicationAndWaitUntilRunning(streams);
final List<KeyValue<Long, Long>> firstBurstOfData = prepareData(0L, 5L, 0L);
final List<KeyValue<Long, Long>> secondBurstOfData = prepareData(5L, 8L, 0L);
@@ -380,7 +380,7 @@ public class EosIntegrationTest {
// after fail over, we should read 40 committed records (even if 50 record got written)
try (final KafkaStreams streams = getKafkaStreams("dummy", false, "appDir", 2, eosConfig, MAX_POLL_INTERVAL_MS)) {
- startKafkaStreamsAndWaitForRunningState(streams, MAX_WAIT_TIME_MS);
+ startApplicationAndWaitUntilRunning(streams);
final List<KeyValue<Long, Long>> committedDataBeforeFailure = prepareData(0L, 10L, 0L, 1L);
final List<KeyValue<Long, Long>> uncommittedDataBeforeFailure = prepareData(10L, 15L, 0L, 1L);
@@ -488,7 +488,7 @@ public class EosIntegrationTest {
// We need more processing time under "with state" situation, so increasing the max.poll.interval.ms
// to avoid unexpected rebalance during test, which will cause unexpected fail over triggered
try (final KafkaStreams streams = getKafkaStreams("dummy", true, "appDir", 2, eosConfig, 3 * MAX_POLL_INTERVAL_MS)) {
- startKafkaStreamsAndWaitForRunningState(streams, MAX_WAIT_TIME_MS);
+ startApplicationAndWaitUntilRunning(streams);
final List<KeyValue<Long, Long>> committedDataBeforeFailure = prepareData(0L, 10L, 0L, 1L);
final List<KeyValue<Long, Long>> uncommittedDataBeforeFailure = prepareData(10L, 15L, 0L, 1L, 2L, 3L);
@@ -504,7 +504,7 @@ public class EosIntegrationTest {
waitForCondition(
() -> commitRequested.get() == 2, MAX_WAIT_TIME_MS,
- "SteamsTasks did not request commit.");
+ "StreamsTasks did not request commit.");
// expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
//
@@ -607,8 +607,8 @@ public class EosIntegrationTest {
final KafkaStreams streams1 = getKafkaStreams("streams1", false, "appDir1", 1, eosConfig, MAX_POLL_INTERVAL_MS);
final KafkaStreams streams2 = getKafkaStreams("streams2", false, "appDir2", 1, eosConfig, MAX_POLL_INTERVAL_MS)
) {
- startKafkaStreamsAndWaitForRunningState(streams1, MAX_WAIT_TIME_MS);
- startKafkaStreamsAndWaitForRunningState(streams2, MAX_WAIT_TIME_MS);
+ startApplicationAndWaitUntilRunning(streams1);
+ startApplicationAndWaitUntilRunning(streams2);
final List<KeyValue<Long, Long>> committedDataBeforeStall = prepareData(0L, 10L, 0L, 1L);
final List<KeyValue<Long, Long>> uncommittedDataBeforeStall = prepareData(10L, 15L, 0L, 1L);
@@ -626,7 +626,7 @@ public class EosIntegrationTest {
waitForCondition(
() -> commitRequested.get() == 2, MAX_WAIT_TIME_MS,
- "SteamsTasks did not request commit.");
+ "StreamsTasks did not request commit.");
// expected end state per output partition (C == COMMIT; A == ABORT; ---> indicate the changes):
//
@@ -757,11 +757,11 @@ public class EosIntegrationTest {
try (final KafkaStreams streams = getKafkaStreams("streams", true, "appDir", 1, eosConfig, MAX_POLL_INTERVAL_MS)) {
writeInputData(writtenData);
- startKafkaStreamsAndWaitForRunningState(streams, MAX_WAIT_TIME_MS);
+ startApplicationAndWaitUntilRunning(streams);
waitForCondition(
() -> commitRequested.get() == 2, MAX_WAIT_TIME_MS,
- "SteamsTasks did not request commit.");
+ "StreamsTasks did not request commit.");
final List<KeyValue<Long, Long>> committedRecords = readResult(SINGLE_PARTITION_OUTPUT_TOPIC, writtenData.size(), CONSUMER_GROUP_ID);
@@ -866,7 +866,7 @@ public class EosIntegrationTest {
this.context = context;
if (withState) {
- state = (KeyValueStore<Long, Long>) context.getStateStore(storeName);
+ state = context.getStateStore(storeName);
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java
index 02111728645..587c303145d 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/InternalTopicIntegrationTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams.integration;
-import java.time.Duration;
import kafka.utils.MockTime;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClientConfig;
@@ -52,6 +51,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import java.io.IOException;
+import java.time.Duration;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -62,7 +62,6 @@ import java.util.concurrent.TimeUnit;
import static java.time.Duration.ofMillis;
import static java.time.Duration.ofSeconds;
-import static java.util.Collections.singletonList;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForCompletion;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -150,7 +149,7 @@ public class InternalTopicIntegrationTest {
}
/*
- * This test just ensures that that the assignor does not get stuck during partition number resolution
+ * This test just ensures that the assignor does not get stuck during partition number resolution
* for internal repartition topics. See KAFKA-10689
*/
@Test
@@ -176,12 +175,13 @@ public class InternalTopicIntegrationTest {
(x, y) -> x + y
);
- final KafkaStreams streams = new KafkaStreams(streamsBuilder.build(), streamsProp);
- startApplicationAndWaitUntilRunning(singletonList(streams), Duration.ofSeconds(60));
+ try (final KafkaStreams streams = new KafkaStreams(streamsBuilder.build(), streamsProp)) {
+ startApplicationAndWaitUntilRunning(streams);
+ }
}
@Test
- public void shouldCompactTopicsForKeyValueStoreChangelogs() {
+ public void shouldCompactTopicsForKeyValueStoreChangelogs() throws Exception {
final String appID = APP_ID + "-compact";
streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
@@ -195,19 +195,19 @@ public class InternalTopicIntegrationTest {
.groupBy(MockMapper.selectValueMapper())
.count(Materialized.as("Counts"));
- final KafkaStreams streams = new KafkaStreams(builder.build(), streamsProp);
- streams.start();
+ try (final KafkaStreams streams = new KafkaStreams(builder.build(), streamsProp)) {
+ startApplicationAndWaitUntilRunning(streams);
- //
- // Step 2: Produce some input data to the input topic.
- //
- produceData(Arrays.asList("hello", "world", "world", "hello world"));
+ //
+ // Step 2: Produce some input data to the input topic.
+ //
+ produceData(Arrays.asList("hello", "world", "world", "hello world"));
- //
- // Step 3: Verify the state changelog topics are compact
- //
- waitForCompletion(streams, 2, 30000L);
- streams.close();
+ //
+ // Step 3: Verify the state changelog topics are compact
+ //
+ waitForCompletion(streams, 2, 30000L);
+ }
final Properties changelogProps = getTopicProperties(ProcessorStateManager.storeChangelogTopic(appID, "Counts", null));
assertEquals(TopicConfig.CLEANUP_POLICY_COMPACT, changelogProps.getProperty(TopicConfig.CLEANUP_POLICY_CONFIG));
@@ -218,7 +218,7 @@ public class InternalTopicIntegrationTest {
}
@Test
- public void shouldCompactAndDeleteTopicsForWindowStoreChangelogs() {
+ public void shouldCompactAndDeleteTopicsForWindowStoreChangelogs() throws Exception {
final String appID = APP_ID + "-compact-delete";
streamsProp.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
@@ -235,19 +235,20 @@ public class InternalTopicIntegrationTest {
.windowedBy(TimeWindows.of(ofSeconds(1L)).grace(ofMillis(0L)))
.count(Materialized.<String, Long, WindowStore<Bytes, byte[]>>as("CountWindows").withRetention(ofSeconds(2L)));
- final KafkaStreams streams = new KafkaStreams(builder.build(), streamsProp);
- streams.start();
+ try (final KafkaStreams streams = new KafkaStreams(builder.build(), streamsProp)) {
+ startApplicationAndWaitUntilRunning(streams);
- //
- // Step 2: Produce some input data to the input topic.
- //
- produceData(Arrays.asList("hello", "world", "world", "hello world"));
+ //
+ // Step 2: Produce some input data to the input topic.
+ //
+ produceData(Arrays.asList("hello", "world", "world", "hello world"));
+
+ //
+ // Step 3: Verify the state changelog topics are compact
+ //
+ waitForCompletion(streams, 2, 30000L);
+ }
- //
- // Step 3: Verify the state changelog topics are compact
- //
- waitForCompletion(streams, 2, 30000L);
- streams.close();
final Properties properties = getTopicProperties(ProcessorStateManager.storeChangelogTopic(appID, "CountWindows", null));
final List<String> policies = Arrays.asList(properties.getProperty(TopicConfig.CLEANUP_POLICY_CONFIG).split(","));
assertEquals(2, policies.size());
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest.java
index fbd2c767d87..ad2901fda41 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest.java
@@ -17,17 +17,6 @@
package org.apache.kafka.streams.integration;
-import static java.time.Duration.ofSeconds;
-import static java.util.Arrays.asList;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
-import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForApplicationState;
-
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.core.IsEqual.equalTo;
-import static org.junit.Assert.assertEquals;
-
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
@@ -72,6 +61,16 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.TestInfo;
+import static java.time.Duration.ofSeconds;
+import static java.util.Arrays.asList;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForApplicationState;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.core.IsEqual.equalTo;
+
@Timeout(600)
@Tag("integration")
public class KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest {
@@ -231,7 +230,7 @@ public class KTableKTableForeignKeyInnerJoinCustomPartitionerIntegrationTest {
OUTPUT,
expectedResult.size()));
- assertEquals(expectedResult, result);
+ assertThat(expectedResult, equalTo(result));
}
private Properties getStreamsConfig(final String testName) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java
index 4f7de4397db..00da6a89e18 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/KafkaStreamsCloseOptionsIntegrationTest.java
@@ -57,7 +57,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Properties;
-import static java.util.Collections.singletonList;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForEmptyConsumerGroup;
@Category({IntegrationTest.class})
@@ -162,7 +161,7 @@ public class KafkaStreamsCloseOptionsIntegrationTest {
// RUN
streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(singletonList(streams), Duration.ofSeconds(30));
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
streams.close(new CloseOptions().leaveGroup(true).timeout(Duration.ofSeconds(30)));
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/LagFetchIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/LagFetchIntegrationTest.java
index 363fed3b8e6..ce21132eebc 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/LagFetchIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/LagFetchIntegrationTest.java
@@ -56,7 +56,6 @@ import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -285,7 +284,7 @@ public class LagFetchIntegrationTest {
"Should see empty lag map before streams is started.");
// Get the instance to fully catch up and reach RUNNING state
- startApplicationAndWaitUntilRunning(Collections.singletonList(streams), Duration.ofSeconds(60));
+ startApplicationAndWaitUntilRunning(streams);
IntegrationTestUtils.waitUntilMinValuesRecordsReceived(
consumerConfiguration,
outputTopicName,
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java
index 9d4f270ed38..6c4908e1bea 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java
@@ -57,7 +57,6 @@ import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.internals.StreamsMetadataImpl;
import org.apache.kafka.streams.utils.UniqueTopicSerdeScope;
import org.apache.kafka.test.IntegrationTest;
-import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import java.util.Arrays;
@@ -107,8 +106,6 @@ import static java.util.Collections.singletonList;
@Timeout(600)
@Category(IntegrationTest.class)
public class NamedTopologyIntegrationTest {
- private static final Duration STARTUP_TIMEOUT = Duration.ofSeconds(45);
-
public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(1);
private static final String TOPOLOGY_1 = "topology-1";
@@ -291,8 +288,9 @@ public class NamedTopologyIntegrationTest {
(value1, value2) -> String.valueOf(value1 + value2),
Materialized.with(null, serdeScope.decorateSerde(Serdes.String(), props, false)));
- streams.start(asList(fkjBuilder.build(), countBuilder.build()));
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(fkjBuilder.build());
+ streams.addNamedTopology(countBuilder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
final String countTopicPrefix = TOPIC_PREFIX + "-" + countTopologyName;
final String fkjTopicPrefix = TOPIC_PREFIX + "-" + fkjTopologyName;
@@ -319,8 +317,8 @@ public class NamedTopologyIntegrationTest {
.groupByKey()
.count(ROCKSDB_STORE)
.toStream().to(OUTPUT_STREAM_1);
- streams.start(topology1Builder.build());
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
final List<KeyValue<String, Long>> results = waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3);
assertThat(results, equalTo(COUNT_OUTPUT_DATA));
@@ -334,8 +332,10 @@ public class NamedTopologyIntegrationTest {
topology1Builder.stream(INPUT_STREAM_1).groupBy((k, v) -> k).count(ROCKSDB_STORE).toStream().to(OUTPUT_STREAM_1);
topology2Builder.stream(INPUT_STREAM_2).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_2);
topology3Builder.stream(INPUT_STREAM_3).groupBy((k, v) -> k).count(ROCKSDB_STORE).toStream().to(OUTPUT_STREAM_3);
- streams.start(asList(topology1Builder.build(), topology2Builder.build(), topology3Builder.build()));
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams.addNamedTopology(topology2Builder.build());
+ streams.addNamedTopology(topology3Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_2, 3), equalTo(COUNT_OUTPUT_DATA));
@@ -364,7 +364,7 @@ public class NamedTopologyIntegrationTest {
assertThat(streams.getTopologyByName(TOPOLOGY_1), is(Optional.empty()));
streams.addNamedTopology(topology1Builder.build());
streams.addNamedTopology(topology2Builder.build());
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(singletonList(streams), Duration.ofSeconds(15));
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, SINGLE_PARTITION_OUTPUT_STREAM, 3), equalTo(COUNT_OUTPUT_DATA));
@@ -403,7 +403,7 @@ public class NamedTopologyIntegrationTest {
topology2Builder2.stream(SINGLE_PARTITION_INPUT_STREAM).groupByKey().count(Materialized.as(topology2Store)).toStream().to(SINGLE_PARTITION_OUTPUT_STREAM);
streams2.start(asList(topology1Builder2.build(), topology2Builder2.build()));
- waitForApplicationState(asList(streams, streams2), State.RUNNING, STARTUP_TIMEOUT);
+ waitForApplicationState(asList(streams, streams2), State.RUNNING, Duration.ofSeconds(60));
verifyMetadataForTopology(
TOPOLOGY_1,
@@ -445,8 +445,8 @@ public class NamedTopologyIntegrationTest {
public void shouldAddNamedTopologyToRunningApplicationWithSingleInitialNamedTopology() throws Exception {
topology1Builder.stream(INPUT_STREAM_1).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_1);
topology2Builder.stream(INPUT_STREAM_2).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_2);
- streams.start(topology1Builder.build());
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
streams.addNamedTopology(topology2Builder.build()).all().get();
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
@@ -458,8 +458,9 @@ public class NamedTopologyIntegrationTest {
topology1Builder.stream(INPUT_STREAM_1).groupBy((k, v) -> k).count(ROCKSDB_STORE).toStream().to(OUTPUT_STREAM_1);
topology2Builder.stream(INPUT_STREAM_2).groupBy((k, v) -> k).count(ROCKSDB_STORE).toStream().to(OUTPUT_STREAM_2);
topology3Builder.stream(INPUT_STREAM_3).groupBy((k, v) -> k).count(ROCKSDB_STORE).toStream().to(OUTPUT_STREAM_3);
- streams.start(asList(topology1Builder.build(), topology2Builder.build()));
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams.addNamedTopology(topology2Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
streams.addNamedTopology(topology3Builder.build()).all().get();
@@ -477,9 +478,9 @@ public class NamedTopologyIntegrationTest {
topology2Builder.stream(INPUT_STREAM_2).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_2);
topology2Builder2.stream(INPUT_STREAM_2).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_2);
- streams.start(topology1Builder.build());
- streams2.start(topology1Builder2.build());
- waitForApplicationState(asList(streams, streams2), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams2.addNamedTopology(topology1Builder2.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(asList(streams, streams2));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
@@ -497,9 +498,9 @@ public class NamedTopologyIntegrationTest {
topology1Builder.stream(INPUT_STREAM_1).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_1);
topology1Builder2.stream(INPUT_STREAM_1).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_1);
- streams.start(topology1Builder.build());
- streams2.start(topology1Builder2.build());
- waitForApplicationState(asList(streams, streams2), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams2.addNamedTopology(topology1Builder2.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(asList(streams, streams2));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
@@ -550,8 +551,9 @@ public class NamedTopologyIntegrationTest {
topology2Builder.stream(DELAYED_INPUT_STREAM_2).map((k, v) -> {
throw new IllegalStateException("Should not process any records for removed topology-2");
});
- streams.start(asList(topology1Builder.build(), topology2Builder.build()));
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams.addNamedTopology(topology2Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
streams.removeNamedTopology("topology-2").all().get();
@@ -574,8 +576,8 @@ public class NamedTopologyIntegrationTest {
final KStream<String, Long> inputStream1 = topology1Builder.stream(INPUT_STREAM_1);
inputStream1.groupByKey().count().toStream().to(COUNT_OUTPUT);
inputStream1.groupByKey().reduce(Long::sum).toStream().to(SUM_OUTPUT);
- streams.start(singletonList(topology1Builder.build()));
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, COUNT_OUTPUT, 3), equalTo(COUNT_OUTPUT_DATA));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, SUM_OUTPUT, 3), equalTo(SUM_OUTPUT_DATA));
@@ -604,8 +606,10 @@ public class NamedTopologyIntegrationTest {
topology1Builder.stream(Pattern.compile(INPUT_STREAM_1)).groupBy((k, v) -> k).count().toStream().to(OUTPUT_STREAM_1);
topology2Builder.stream(Pattern.compile(INPUT_STREAM_2)).groupBy((k, v) -> k).count().toStream().to(OUTPUT_STREAM_2);
topology3Builder.stream(Pattern.compile(INPUT_STREAM_3)).groupBy((k, v) -> k).count().toStream().to(OUTPUT_STREAM_3);
- streams.start(asList(topology1Builder.build(), topology2Builder.build(), topology3Builder.build()));
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams.addNamedTopology(topology2Builder.build());
+ streams.addNamedTopology(topology3Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_2, 3), equalTo(COUNT_OUTPUT_DATA));
@@ -617,8 +621,10 @@ public class NamedTopologyIntegrationTest {
topology1Builder.stream(INPUT_STREAM_1).groupBy((k, v) -> k).count().toStream().to(OUTPUT_STREAM_1);
topology2Builder.stream(Pattern.compile(INPUT_STREAM_2)).groupBy((k, v) -> k).count().toStream().to(OUTPUT_STREAM_2);
topology3Builder.stream(Pattern.compile(INPUT_STREAM_3)).groupBy((k, v) -> k).count().toStream().to(OUTPUT_STREAM_3);
- streams.start(asList(topology1Builder.build(), topology2Builder.build(), topology3Builder.build()));
- waitForApplicationState(singletonList(streams), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams.addNamedTopology(topology2Builder.build());
+ streams.addNamedTopology(topology3Builder.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_2, 3), equalTo(COUNT_OUTPUT_DATA));
@@ -714,9 +720,10 @@ public class NamedTopologyIntegrationTest {
streams.setUncaughtExceptionHandler(handler);
streams2.setUncaughtExceptionHandler(handler);
- streams.start(topology1Builder.build());
- streams2.start(topology1Builder2.build());
- waitForApplicationState(asList(streams, streams2), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams2.addNamedTopology(topology1Builder2.build());
+
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(asList(streams, streams2));
assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3), equalTo(COUNT_OUTPUT_DATA));
topology2Builder.stream(NEW_STREAM).groupBy((k, v) -> k).count(IN_MEMORY_STORE).toStream().to(OUTPUT_STREAM_2);
@@ -772,9 +779,9 @@ public class NamedTopologyIntegrationTest {
streams.setUncaughtExceptionHandler(handler);
streams2.setUncaughtExceptionHandler(handler);
- streams.start(topology1Builder.build());
- streams2.start(topology1Builder2.build());
- waitForApplicationState(asList(streams, streams2), State.RUNNING, STARTUP_TIMEOUT);
+ streams.addNamedTopology(topology1Builder.build());
+ streams2.addNamedTopology(topology1Builder2.build());
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(asList(streams, streams2));
retryOnExceptionWithTimeout(() -> {
final Throwable error = handler.nextError(TOPOLOGY_1);
@@ -840,10 +847,10 @@ public class NamedTopologyIntegrationTest {
streams.addNamedTopology(builder.build());
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(streams);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(streams);
IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
DELAYED_INPUT_STREAM_2,
- Arrays.asList(
+ singletonList(
new KeyValue<>(1, "A")
),
TestUtils.producerConfig(
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java
index 5ebe2d27575..9f6e1d213b0 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/OptimizedKTableIntegrationTest.java
@@ -121,57 +121,62 @@ public class OptimizedKTableIntegrationTest {
final KafkaStreams kafkaStreams2 = createKafkaStreams(builder, streamsConfiguration(testInfo));
final List<KafkaStreams> kafkaStreamsList = Arrays.asList(kafkaStreams1, kafkaStreams2);
- startApplicationAndWaitUntilRunning(kafkaStreamsList, Duration.ofSeconds(60));
+ try {
+ startApplicationAndWaitUntilRunning(kafkaStreamsList, Duration.ofSeconds(60));
+
+ produceValueRange(key, 0, batch1NumMessages);
+
+ // Assert that all messages in the first batch were processed in a timely manner
+ assertThat(semaphore.tryAcquire(batch1NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));
+
+ final AtomicReference<ReadOnlyKeyValueStore<Integer, Integer>> newActiveStore = new AtomicReference<>(null);
+ TestUtils.retryOnExceptionWithTimeout(() -> {
+ final ReadOnlyKeyValueStore<Integer, Integer> store1 = IntegrationTestUtils.getStore(TABLE_NAME, kafkaStreams1, QueryableStoreTypes.keyValueStore());
+ final ReadOnlyKeyValueStore<Integer, Integer> store2 = IntegrationTestUtils.getStore(TABLE_NAME, kafkaStreams2, QueryableStoreTypes.keyValueStore());
+
+ final KeyQueryMetadata keyQueryMetadata = kafkaStreams1.queryMetadataForKey(TABLE_NAME, key, (topic, somekey, value, numPartitions) -> 0);
+
+ try {
+ // Assert that the current value in store reflects all messages being processed
+ if ((keyQueryMetadata.activeHost().port() % 2) == 1) {
+ assertThat(store1.get(key), is(equalTo(batch1NumMessages - 1)));
+ kafkaStreams1.close();
+ newActiveStore.set(store2);
+ } else {
+ assertThat(store2.get(key), is(equalTo(batch1NumMessages - 1)));
+ kafkaStreams2.close();
+ newActiveStore.set(store1);
+ }
+ } catch (final InvalidStateStoreException e) {
+ LOG.warn("Detected an unexpected rebalance during test. Retrying if possible.", e);
+ throw e;
+ } catch (final Throwable t) {
+ LOG.error("Caught non-retriable exception in test. Exiting.", t);
+ throw new NoRetryException(t);
+ }
+ });
- produceValueRange(key, 0, batch1NumMessages);
+ // Wait for failover
+ TestUtils.retryOnExceptionWithTimeout(60 * 1000, 100, () -> {
+ // Assert that after failover we have recovered to the last store write
+ assertThat(newActiveStore.get().get(key), is(equalTo(batch1NumMessages - 1)));
+ });
- // Assert that all messages in the first batch were processed in a timely manner
- assertThat(semaphore.tryAcquire(batch1NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));
+ final int totalNumMessages = batch1NumMessages + batch2NumMessages;
- final AtomicReference<ReadOnlyKeyValueStore<Integer, Integer>> newActiveStore = new AtomicReference<>(null);
- TestUtils.retryOnExceptionWithTimeout(() -> {
- final ReadOnlyKeyValueStore<Integer, Integer> store1 = IntegrationTestUtils.getStore(TABLE_NAME, kafkaStreams1, QueryableStoreTypes.keyValueStore());
- final ReadOnlyKeyValueStore<Integer, Integer> store2 = IntegrationTestUtils.getStore(TABLE_NAME, kafkaStreams2, QueryableStoreTypes.keyValueStore());
+ produceValueRange(key, batch1NumMessages, totalNumMessages);
- final KeyQueryMetadata keyQueryMetadata = kafkaStreams1.queryMetadataForKey(TABLE_NAME, key, (topic, somekey, value, numPartitions) -> 0);
+ // Assert that all messages in the second batch were processed in a timely manner
+ assertThat(semaphore.tryAcquire(batch2NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));
- try {
+ TestUtils.retryOnExceptionWithTimeout(60 * 1000, 100, () -> {
// Assert that the current value in store reflects all messages being processed
- if ((keyQueryMetadata.activeHost().port() % 2) == 1) {
- assertThat(store1.get(key), is(equalTo(batch1NumMessages - 1)));
- kafkaStreams1.close();
- newActiveStore.set(store2);
- } else {
- assertThat(store2.get(key), is(equalTo(batch1NumMessages - 1)));
- kafkaStreams2.close();
- newActiveStore.set(store1);
- }
- } catch (final InvalidStateStoreException e) {
- LOG.warn("Detected an unexpected rebalance during test. Retrying if possible.", e);
- throw e;
- } catch (final Throwable t) {
- LOG.error("Caught non-retriable exception in test. Exiting.", t);
- throw new NoRetryException(t);
- }
- });
-
- // Wait for failover
- TestUtils.retryOnExceptionWithTimeout(60 * 1000, 100, () -> {
- // Assert that after failover we have recovered to the last store write
- assertThat(newActiveStore.get().get(key), is(equalTo(batch1NumMessages - 1)));
- });
-
- final int totalNumMessages = batch1NumMessages + batch2NumMessages;
-
- produceValueRange(key, batch1NumMessages, totalNumMessages);
-
- // Assert that all messages in the second batch were processed in a timely manner
- assertThat(semaphore.tryAcquire(batch2NumMessages, 60, TimeUnit.SECONDS), is(equalTo(true)));
-
- TestUtils.retryOnExceptionWithTimeout(60 * 1000, 100, () -> {
- // Assert that the current value in store reflects all messages being processed
- assertThat(newActiveStore.get().get(key), is(equalTo(totalNumMessages - 1)));
- });
+ assertThat(newActiveStore.get().get(key), is(equalTo(totalNumMessages - 1)));
+ });
+ } finally {
+ kafkaStreams1.close();
+ kafkaStreams2.close();
+ }
}
private void produceValueRange(final int key, final int start, final int endExclusive) {
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java
index d0f04a1265e..92d63219671 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java
@@ -110,7 +110,6 @@ import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.st
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForApplicationState;
import static org.apache.kafka.streams.state.QueryableStoreTypes.keyValueStore;
import static org.apache.kafka.streams.state.QueryableStoreTypes.sessionStore;
-import static org.apache.kafka.test.StreamsTestUtils.startKafkaStreamsAndWaitForRunningState;
import static org.apache.kafka.test.TestUtils.retryOnExceptionWithTimeout;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
@@ -554,12 +553,13 @@ public class QueryableStateIntegrationTest {
listeners.add(listener);
streamsList.add(streams);
}
- startApplicationAndWaitUntilRunning(streamsList, Duration.ofSeconds(60));
-
- final Set<String> stores = mkSet(storeName + "-" + streamThree, windowStoreName + "-" + streamThree);
- verifyOffsetLagFetch(streamsList, stores, Arrays.asList(4, 4));
try {
+ startApplicationAndWaitUntilRunning(streamsList, Duration.ofSeconds(60));
+
+ final Set<String> stores = mkSet(storeName + "-" + streamThree, windowStoreName + "-" + streamThree);
+ verifyOffsetLagFetch(streamsList, stores, Arrays.asList(4, 4));
+
waitUntilAtLeastNumRecordProcessed(outputTopicThree, 1);
for (int i = 0; i < streamsList.size(); i++) {
@@ -656,10 +656,11 @@ public class QueryableStateIntegrationTest {
listeners.add(listener);
streamsList.add(streams);
}
- startApplicationAndWaitUntilRunning(streamsList, ofSeconds(60));
- verifyOffsetLagFetch(streamsList, stores, Arrays.asList(8, 8));
try {
+ startApplicationAndWaitUntilRunning(streamsList, ofSeconds(60));
+ verifyOffsetLagFetch(streamsList, stores, Arrays.asList(8, 8));
+
waitUntilAtLeastNumRecordProcessed(outputTopicThree, 1);
// Ensure each thread can serve all keys by itself; i.e standby replication works.
@@ -766,7 +767,7 @@ public class QueryableStateIntegrationTest {
t2.toStream().to(outputTopic);
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
waitUntilAtLeastNumRecordProcessed(outputTopic, 1);
@@ -833,7 +834,7 @@ public class QueryableStateIntegrationTest {
.to(outputTopic, Produced.with(Serdes.String(), Serdes.Long()));
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
waitUntilAtLeastNumRecordProcessed(outputTopic, 5);
@@ -890,7 +891,7 @@ public class QueryableStateIntegrationTest {
.to(outputTopic, Produced.with(Serdes.String(), Serdes.Long()));
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
waitUntilAtLeastNumRecordProcessed(outputTopic, 5);
@@ -942,7 +943,7 @@ public class QueryableStateIntegrationTest {
t3.toStream().to(outputTopic, Produced.with(Serdes.String(), Serdes.Long()));
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
waitUntilAtLeastNumRecordProcessed(outputTopic, 1);
@@ -1003,7 +1004,7 @@ public class QueryableStateIntegrationTest {
.windowedBy(TimeWindows.of(ofMillis(WINDOW_SIZE)))
.count(Materialized.as(windowStoreName));
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
waitUntilAtLeastNumRecordProcessed(outputTopic, 1);
@@ -1030,7 +1031,7 @@ public class QueryableStateIntegrationTest {
final String storeName = "count-by-key";
stream.groupByKey().count(Materialized.as(storeName));
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
final KeyValue<String, String> hello = KeyValue.pair("hello", "hello");
IntegrationTestUtils.produceKeyValuesSynchronously(
@@ -1058,7 +1059,7 @@ public class QueryableStateIntegrationTest {
// start again, and since it may take time to restore we wait for it to transit to RUNNING a bit longer
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
- startKafkaStreamsAndWaitForRunningState(kafkaStreams, maxWaitMs);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
// make sure we never get any value other than 8 for hello
TestUtils.waitForCondition(
@@ -1101,8 +1102,7 @@ public class QueryableStateIntegrationTest {
kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration);
kafkaStreams.setUncaughtExceptionHandler((t, e) -> failed.set(true));
- // since we start with two threads, wait for a bit longer for both of them to transit to running
- startKafkaStreamsAndWaitForRunningState(kafkaStreams, 30000);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
IntegrationTestUtils.produceKeyValuesSynchronously(
streamOne,
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RangeQueryIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RangeQueryIntegrationTest.java
index c67696b2cad..7ad97b151d7 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/RangeQueryIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/RangeQueryIntegrationTest.java
@@ -17,7 +17,6 @@
package org.apache.kafka.streams.integration;
-import java.util.Collections;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.StringSerializer;
@@ -52,7 +51,6 @@ import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import java.io.IOException;
-import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
@@ -181,7 +179,7 @@ public class RangeQueryIntegrationTest {
builder.table(inputStream, stateStoreConfig);
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), STREAMS_CONFIG)) {
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(Collections.singletonList(kafkaStreams), Duration.ofSeconds(60));
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
writeInputData();
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
index 4dab0f3baec..ccebed8d5c1 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/RegexSourceIntegrationTest.java
@@ -57,7 +57,6 @@ import org.junit.jupiter.api.TestInfo;
import org.junit.jupiter.api.Timeout;
import java.io.IOException;
-import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -223,7 +222,7 @@ public class RegexSourceIntegrationTest {
assertThat(topology.describe().subtopologies().size(), greaterThan(1));
streams = new KafkaStreams(topology, streamsConfiguration);
- startApplicationAndWaitUntilRunning(Collections.singletonList(streams), Duration.ofSeconds(30));
+ startApplicationAndWaitUntilRunning(streams);
CLUSTER.createTopic(topic2);
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
index 09255268333..9c47a330f04 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java
@@ -434,27 +434,32 @@ public class RestoreIntegrationTest {
props1.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
props1.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId + "-1").getPath());
purgeLocalStreamsState(props1);
- final KafkaStreams client1 = new KafkaStreams(builder.build(), props1);
+ final KafkaStreams streams1 = new KafkaStreams(builder.build(), props1);
final Properties props2 = props(stateUpdaterEnabled);
props2.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);
props2.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory(appId + "-2").getPath());
purgeLocalStreamsState(props2);
- final KafkaStreams client2 = new KafkaStreams(builder.build(), props2);
+ final KafkaStreams streams2 = new KafkaStreams(builder.build(), props2);
final Set<KafkaStreams.State> transitionedStates1 = Collections.newSetFromMap(new ConcurrentHashMap<>());
final Set<KafkaStreams.State> transitionedStates2 = Collections.newSetFromMap(new ConcurrentHashMap<>());
final TrackingStateRestoreListener restoreListener = new TrackingStateRestoreListener();
- client1.setGlobalStateRestoreListener(restoreListener);
- client1.setStateListener((newState, oldState) -> transitionedStates1.add(newState));
- client2.setStateListener((newState, oldState) -> transitionedStates2.add(newState));
-
- startApplicationAndWaitUntilRunning(asList(client1, client2), Duration.ofSeconds(60));
-
- waitForCompletion(client1, 1, 30 * 1000L);
- waitForCompletion(client2, 1, 30 * 1000L);
- waitForStandbyCompletion(client1, 1, 30 * 1000L);
- waitForStandbyCompletion(client2, 1, 30 * 1000L);
+ streams1.setGlobalStateRestoreListener(restoreListener);
+ streams1.setStateListener((newState, oldState) -> transitionedStates1.add(newState));
+ streams2.setStateListener((newState, oldState) -> transitionedStates2.add(newState));
+
+ try {
+ startApplicationAndWaitUntilRunning(asList(streams1, streams2), Duration.ofSeconds(60));
+
+ waitForCompletion(streams1, 1, 30 * 1000L);
+ waitForCompletion(streams2, 1, 30 * 1000L);
+ waitForStandbyCompletion(streams1, 1, 30 * 1000L);
+ waitForStandbyCompletion(streams2, 1, 30 * 1000L);
+ } catch (final Exception e) {
+ streams1.close();
+ streams2.close();
+ }
// Sometimes the store happens to have already been closed sometime during startup, so just keep track
// of where it started and make sure it doesn't happen more times from there
@@ -463,21 +468,23 @@ public class RestoreIntegrationTest {
transitionedStates1.clear();
transitionedStates2.clear();
- client2.close();
- waitForTransitionTo(transitionedStates2, State.NOT_RUNNING, Duration.ofSeconds(60));
- waitForTransitionTo(transitionedStates1, State.REBALANCING, Duration.ofSeconds(60));
- waitForTransitionTo(transitionedStates1, State.RUNNING, Duration.ofSeconds(60));
-
- waitForCompletion(client1, 1, 30 * 1000L);
- waitForStandbyCompletion(client1, 1, 30 * 1000L);
-
- assertThat(restoreListener.totalNumRestored(), CoreMatchers.equalTo(initialNunRestoredCount));
-
- // After stopping instance 2 and letting instance 1 take over its tasks, we should have closed just two stores
- // total: the active and standby tasks on instance 2
- assertThat(CloseCountingInMemoryStore.numStoresClosed(), equalTo(initialStoreCloseCount + 2));
-
- client1.close();
+ try {
+ streams2.close();
+ waitForTransitionTo(transitionedStates2, State.NOT_RUNNING, Duration.ofSeconds(60));
+ waitForTransitionTo(transitionedStates1, State.REBALANCING, Duration.ofSeconds(60));
+ waitForTransitionTo(transitionedStates1, State.RUNNING, Duration.ofSeconds(60));
+
+ waitForCompletion(streams1, 1, 30 * 1000L);
+ waitForStandbyCompletion(streams1, 1, 30 * 1000L);
+
+ assertThat(restoreListener.totalNumRestored(), CoreMatchers.equalTo(initialNunRestoredCount));
+
+ // After stopping instance 2 and letting instance 1 take over its tasks, we should have closed just two stores
+ // total: the active and standby tasks on instance 2
+ assertThat(CloseCountingInMemoryStore.numStoresClosed(), equalTo(initialStoreCloseCount + 2));
+ } finally {
+ streams1.close();
+ }
waitForTransitionTo(transitionedStates1, State.NOT_RUNNING, Duration.ofSeconds(60));
assertThat(CloseCountingInMemoryStore.numStoresClosed(), CoreMatchers.equalTo(initialStoreCloseCount + 4));
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java
index 2d667839fae..02d34b51fb7 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/RocksDBMetricsIntegrationTest.java
@@ -37,7 +37,6 @@ import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.test.IntegrationTest;
-import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.AfterClass;
@@ -64,6 +63,8 @@ import java.util.Properties;
import java.util.stream.Collectors;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
+
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
@@ -232,7 +233,7 @@ public class RocksDBMetricsIntegrationTest {
kafkaStreams.cleanUp();
produceRecords();
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams, TIMEOUT);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
metricsVerifier.verify(kafkaStreams, "rocksdb-state-id");
metricsVerifier.verify(kafkaStreams, "rocksdb-window-state-id");
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java
index 2ac2c2bddee..eaa24e8005a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java
@@ -229,7 +229,7 @@ public class StandbyTaskEOSIntegrationTest {
streamInstanceTwo = buildWithDeduplicationTopology(base + "-2");
// start first instance and wait for processing
- startApplicationAndWaitUntilRunning(Collections.singletonList(streamInstanceOne), Duration.ofSeconds(30));
+ startApplicationAndWaitUntilRunning(streamInstanceOne);
IntegrationTestUtils.waitUntilMinRecordsReceived(
TestUtils.consumerConfig(
CLUSTER.bootstrapServers(),
@@ -241,7 +241,7 @@ public class StandbyTaskEOSIntegrationTest {
);
// start second instance and wait for standby replication
- startApplicationAndWaitUntilRunning(Collections.singletonList(streamInstanceTwo), Duration.ofSeconds(30));
+ startApplicationAndWaitUntilRunning(streamInstanceTwo);
waitForCondition(
() -> streamInstanceTwo.store(
StoreQueryParameters.fromNameAndType(
@@ -297,10 +297,7 @@ public class StandbyTaskEOSIntegrationTest {
// "restart" first client and wait for standby recovery
// (could actually also be active, but it does not matter as long as we enable "state stores"
- startApplicationAndWaitUntilRunning(
- Collections.singletonList(streamInstanceOneRecovery),
- Duration.ofSeconds(30)
- );
+ startApplicationAndWaitUntilRunning(streamInstanceOneRecovery);
waitForCondition(
() -> streamInstanceOneRecovery.store(
StoreQueryParameters.fromNameAndType(
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java
index 5799383bcde..8b5b1ebd62a 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/StoreQueryIntegrationTest.java
@@ -78,12 +78,12 @@ import static org.apache.kafka.streams.state.QueryableStoreTypes.keyValueStore;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.anyOf;
import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@Timeout(600)
@@ -369,8 +369,8 @@ public class StoreQueryIntegrationTest {
startApplicationAndWaitUntilRunning(kafkaStreamsList, Duration.ofSeconds(60));
- assertTrue(kafkaStreams1.metadataForLocalThreads().size() > 1);
- assertTrue(kafkaStreams2.metadataForLocalThreads().size() > 1);
+ assertThat(kafkaStreams1.metadataForLocalThreads().size(), greaterThan(1));
+ assertThat(kafkaStreams2.metadataForLocalThreads().size(), greaterThan(1));
produceValueRange(key, 0, batch1NumMessages);
@@ -440,8 +440,8 @@ public class StoreQueryIntegrationTest {
kafkaStreams2.start(builder2A.build());
waitForApplicationState(kafkaStreamsList, State.RUNNING, Duration.ofSeconds(60));
- assertTrue(kafkaStreams1.metadataForLocalThreads().size() > 1);
- assertTrue(kafkaStreams2.metadataForLocalThreads().size() > 1);
+ assertThat(kafkaStreams1.metadataForLocalThreads().size(), greaterThan(1));
+ assertThat(kafkaStreams2.metadataForLocalThreads().size(), greaterThan(1));
produceValueRange(key, 0, batch1NumMessages);
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
index 4af333a65a8..09147ce8f19 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/StreamsUncaughtExceptionHandlerIntegrationTest.java
@@ -39,7 +39,6 @@ import org.apache.kafka.streams.kstream.Named;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.internals.KeyValueStoreBuilder;
import org.apache.kafka.test.IntegrationTest;
-import org.apache.kafka.test.StreamsTestUtils;
import org.apache.kafka.test.TestUtils;
import org.junit.After;
import org.junit.AfterClass;
@@ -54,7 +53,6 @@ import org.junit.rules.Timeout;
import java.io.IOException;
import java.time.Duration;
import java.util.ArrayList;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -69,10 +67,12 @@ import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.St
import static org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.purgeLocalStreamsState;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName;
+import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.startApplicationAndWaitUntilRunning;
import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForApplicationState;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.fail;
+import static java.util.Arrays.asList;
@Category(IntegrationTest.class)
@SuppressWarnings("deprecation") //Need to call the old handler, will remove those calls when the old handler is removed
@@ -136,12 +136,12 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
}
@Test
- public void shouldShutdownThreadUsingOldHandler() throws InterruptedException {
+ public void shouldShutdownThreadUsingOldHandler() throws Exception {
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
final AtomicInteger counter = new AtomicInteger(0);
kafkaStreams.setUncaughtExceptionHandler((t, e) -> counter.incrementAndGet());
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
produceMessages(0L, inputTopic, "A");
// should call the UncaughtExceptionHandler in current thread
@@ -156,13 +156,13 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
}
@Test
- public void shouldShutdownClient() throws InterruptedException {
+ public void shouldShutdownClient() throws Exception {
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler"));
kafkaStreams.setUncaughtExceptionHandler(exception -> SHUTDOWN_CLIENT);
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
produceMessages(0L, inputTopic, "A");
waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION);
@@ -172,33 +172,33 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
}
@Test
- public void shouldReplaceThreads() throws InterruptedException {
+ public void shouldReplaceThreads() throws Exception {
testReplaceThreads(2);
}
@Test
- public void shouldReplaceThreadsWithoutJavaHandler() throws InterruptedException {
+ public void shouldReplaceThreadsWithoutJavaHandler() throws Exception {
Thread.setDefaultUncaughtExceptionHandler((t, e) -> fail("exception thrown"));
testReplaceThreads(2);
}
@Test
- public void shouldReplaceSingleThread() throws InterruptedException {
+ public void shouldReplaceSingleThread() throws Exception {
testReplaceThreads(1);
}
@Test
- public void shouldShutdownMultipleThreadApplication() throws InterruptedException {
+ public void shouldShutdownMultipleThreadApplication() throws Exception {
testShutdownApplication(2);
}
@Test
- public void shouldShutdownSingleThreadApplication() throws InterruptedException {
+ public void shouldShutdownSingleThreadApplication() throws Exception {
testShutdownApplication(1);
}
@Test
- public void shouldShutDownClientIfGlobalStreamThreadWantsToReplaceThread() throws InterruptedException {
+ public void shouldShutDownClientIfGlobalStreamThreadWantsToReplaceThread() throws Exception {
builder.addGlobalStore(
new KeyValueStoreBuilder<>(
Stores.persistentKeyValueStore("globalStore"),
@@ -216,7 +216,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler"));
kafkaStreams.setUncaughtExceptionHandler(exception -> REPLACE_THREAD);
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
produceMessages(0L, inputTopic2, "A");
waitForApplicationState(Collections.singletonList(kafkaStreams), KafkaStreams.State.ERROR, DEFAULT_DURATION);
@@ -251,11 +251,11 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
kafkaStreams.setUncaughtExceptionHandler(exception -> StreamThreadExceptionResponse.REPLACE_THREAD);
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
inputTopic,
- Arrays.asList(
+ asList(
new KeyValue<>(1, "A"),
new KeyValue<>(1, "B")
),
@@ -268,7 +268,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
inputTopic2,
- Arrays.asList(
+ asList(
new KeyValue<>(1, "A"),
new KeyValue<>(1, "B")
),
@@ -286,7 +286,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
StringDeserializer.class
),
outputTopic,
- Arrays.asList(
+ asList(
new KeyValue<>(1, "A"),
new KeyValue<>(1, "B")
)
@@ -298,7 +298,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
StringDeserializer.class
),
outputTopic2,
- Arrays.asList(
+ asList(
new KeyValue<>(1, "A"),
new KeyValue<>(1, "B")
)
@@ -335,7 +335,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
}
}
- private void testShutdownApplication(final int numThreads) throws InterruptedException {
+ private void testShutdownApplication(final int numThreads) throws Exception {
properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numThreads);
final Topology topology = builder.build();
@@ -347,17 +347,16 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
kafkaStreams1.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION);
kafkaStreams2.setUncaughtExceptionHandler(exception -> SHUTDOWN_APPLICATION);
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams1);
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams2);
+ startApplicationAndWaitUntilRunning(asList(kafkaStreams1, kafkaStreams2));
produceMessages(0L, inputTopic, "A");
- waitForApplicationState(Arrays.asList(kafkaStreams1, kafkaStreams2), KafkaStreams.State.ERROR, DEFAULT_DURATION);
+ waitForApplicationState(asList(kafkaStreams1, kafkaStreams2), KafkaStreams.State.ERROR, DEFAULT_DURATION);
assertThat(processorValueCollector.size(), equalTo(1));
}
}
- private void testReplaceThreads(final int numThreads) throws InterruptedException {
+ private void testReplaceThreads(final int numThreads) throws Exception {
properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, numThreads);
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
kafkaStreams.setUncaughtExceptionHandler((t, e) -> fail("should not hit old handler"));
@@ -369,7 +368,7 @@ public class StreamsUncaughtExceptionHandlerIntegrationTest {
}
return REPLACE_THREAD;
});
- StreamsTestUtils.startKafkaStreamsAndWaitForRunningState(kafkaStreams);
+ startApplicationAndWaitUntilRunning(kafkaStreams);
produceMessages(0L, inputTopic, "A");
TestUtils.waitForCondition(() -> count.get() == numThreads, "finished replacing threads");
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/TaskMetadataIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/TaskMetadataIntegrationTest.java
index 791ee58ff7b..3c18a7b5fb6 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/TaskMetadataIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/TaskMetadataIntegrationTest.java
@@ -43,7 +43,6 @@ import org.junit.rules.TestName;
import org.junit.rules.Timeout;
import java.io.IOException;
-import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
@@ -75,7 +74,6 @@ public class TaskMetadataIntegrationTest {
public static void closeCluster() {
CLUSTER.stop();
}
- public static final Duration DEFAULT_DURATION = Duration.ofSeconds(30);
@Rule
public TestName testName = new TestName();
@@ -119,7 +117,7 @@ public class TaskMetadataIntegrationTest {
@Test
public void shouldReportCorrectCommittedOffsetInformation() {
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(Collections.singletonList(kafkaStreams), DEFAULT_DURATION);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
final TaskMetadata taskMetadata = getTaskMetadata(kafkaStreams);
assertThat(taskMetadata.committedOffsets().size(), equalTo(1));
final TopicPartition topicPartition = new TopicPartition(inputTopic, 0);
@@ -145,7 +143,7 @@ public class TaskMetadataIntegrationTest {
@Test
public void shouldReportCorrectEndOffsetInformation() {
try (final KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), properties)) {
- IntegrationTestUtils.startApplicationAndWaitUntilRunning(Collections.singletonList(kafkaStreams), DEFAULT_DURATION);
+ IntegrationTestUtils.startApplicationAndWaitUntilRunning(kafkaStreams);
final TaskMetadata taskMetadata = getTaskMetadata(kafkaStreams);
assertThat(taskMetadata.endOffsets().size(), equalTo(1));
final TopicPartition topicPartition = new TopicPartition(inputTopic, 0);
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
index 14478f7b140..ba0807fe3c2 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java
@@ -72,7 +72,6 @@ import java.nio.file.Paths;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
@@ -103,6 +102,7 @@ import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.fail;
+import static java.util.Collections.singletonList;
/**
* Utility functions to make integration testing more convenient.
@@ -960,6 +960,14 @@ public class IntegrationTestUtils {
});
}
+ public static void startApplicationAndWaitUntilRunning(final KafkaStreams streams) throws Exception {
+ startApplicationAndWaitUntilRunning(singletonList(streams));
+ }
+
+ public static void startApplicationAndWaitUntilRunning(final List<KafkaStreams> streamsList) throws Exception {
+ startApplicationAndWaitUntilRunning(streamsList, Duration.ofSeconds(DEFAULT_TIMEOUT));
+ }
+
/**
* Starts the given {@link KafkaStreams} instances and waits for all of them to reach the
* {@link State#RUNNING} state at the same time. Note that states may change between the time
@@ -1099,7 +1107,7 @@ public class IntegrationTestUtils {
final String applicationId) {
try {
final ConsumerGroupDescription groupDescription =
- adminClient.describeConsumerGroups(Collections.singletonList(applicationId))
+ adminClient.describeConsumerGroups(singletonList(applicationId))
.describedGroups()
.get(applicationId)
.get();
@@ -1282,7 +1290,7 @@ public class IntegrationTestUtils {
final long waitTime,
final int maxMessages) {
final List<ConsumerRecord<K, V>> consumerRecords;
- consumer.subscribe(Collections.singletonList(topic));
+ consumer.subscribe(singletonList(topic));
System.out.println("Got assignment:" + consumer.assignment());
final int pollIntervalMs = 100;
consumerRecords = new ArrayList<>();
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java
index 9f1e642a17e..3e7449f9411 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java
@@ -49,7 +49,7 @@ import static org.apache.kafka.streams.StreamsConfig.producerPrefix;
import static org.apache.kafka.test.StreamsTestUtils.TaskBuilder.standbyTask;
import static org.apache.kafka.test.StreamsTestUtils.TaskBuilder.statefulTask;
import static org.apache.kafka.test.StreamsTestUtils.TaskBuilder.statelessTask;
-import static org.apache.kafka.test.StreamsTestUtils.TopologyMetadataBuilder.unamedTopology;
+import static org.apache.kafka.test.StreamsTestUtils.TopologyMetadataBuilder.unnamedTopology;
import static org.apache.kafka.test.TestUtils.waitForCondition;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -72,7 +72,7 @@ class DefaultStateUpdaterTest {
private final static int COMMIT_INTERVAL = 100;
private final static long CALL_TIMEOUT = 1000;
- private final static long VERIFICATION_TIMEOUT = 15000;
+ private final static long VERIFICATION_TIMEOUT = 30000;
private final static TopicPartition TOPIC_PARTITION_A_0 = new TopicPartition("topicA", 0);
private final static TopicPartition TOPIC_PARTITION_A_1 = new TopicPartition("topicA", 1);
private final static TopicPartition TOPIC_PARTITION_B_0 = new TopicPartition("topicB", 0);
@@ -90,7 +90,7 @@ class DefaultStateUpdaterTest {
private final Time time = new MockTime(1L);
private final StreamsConfig config = new StreamsConfig(configProps(COMMIT_INTERVAL));
private final ChangelogReader changelogReader = mock(ChangelogReader.class);
- private final TopologyMetadata topologyMetadata = unamedTopology().build();
+ private final TopologyMetadata topologyMetadata = unnamedTopology().build();
private DefaultStateUpdater stateUpdater = new DefaultStateUpdater("test-state-updater", config, changelogReader, topologyMetadata, time);
@AfterEach
diff --git a/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java b/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java
index 1ba6a1fa0a8..f2528a9a2c9 100644
--- a/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java
+++ b/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java
@@ -22,7 +22,6 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
-import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Windowed;
@@ -47,12 +46,9 @@ import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import static org.apache.kafka.common.metrics.Sensor.RecordingLevel.DEBUG;
-import static org.apache.kafka.test.TestUtils.DEFAULT_MAX_WAIT_MS;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertFalse;
@@ -117,27 +113,6 @@ public final class StreamsTestUtils {
return getStreamsConfig(UUID.randomUUID().toString());
}
- public static void startKafkaStreamsAndWaitForRunningState(final KafkaStreams kafkaStreams) throws InterruptedException {
- startKafkaStreamsAndWaitForRunningState(kafkaStreams, DEFAULT_MAX_WAIT_MS);
- }
-
- public static void startKafkaStreamsAndWaitForRunningState(final KafkaStreams kafkaStreams,
- final long timeoutMs) throws InterruptedException {
- final CountDownLatch countDownLatch = new CountDownLatch(1);
- kafkaStreams.setStateListener((newState, oldState) -> {
- if (newState == KafkaStreams.State.RUNNING) {
- countDownLatch.countDown();
- }
- });
-
- kafkaStreams.start();
- assertThat(
- "KafkaStreams did not transit to RUNNING state within " + timeoutMs + " milli seconds.",
- countDownLatch.await(timeoutMs, TimeUnit.MILLISECONDS),
- equalTo(true)
- );
- }
-
public static <K, V> List<KeyValue<K, V>> toList(final Iterator<KeyValue<K, V>> iterator) {
final List<KeyValue<K, V>> results = new ArrayList<>();
@@ -345,7 +320,7 @@ public final class StreamsTestUtils {
this.topologyMetadata = topologyMetadata;
}
- public static TopologyMetadataBuilder unamedTopology() {
+ public static TopologyMetadataBuilder unnamedTopology() {
final TopologyMetadata topologyMetadata = mock(TopologyMetadata.class);
when(topologyMetadata.isPaused(null)).thenReturn(false);
return new TopologyMetadataBuilder(topologyMetadata);