Increase timeout for waiting for consumer group to be empty

apache · Sep 24, 2024 · 5ebe171 · 5ebe171
1 parent 3633049
commit 5ebe171
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java
@@ -150,7 +150,7 @@ private void prepareConfigs(final String appID) {
 
     protected static final int STREAMS_CONSUMER_TIMEOUT = 2000;
     protected static final int CLEANUP_CONSUMER_TIMEOUT = 2000;
-    protected static final int TIMEOUT_MULTIPLIER = 15;
+    protected static final int TIMEOUT_MULTIPLIER = 30;
 
     void prepareTest(final TestInfo testInfo) throws Exception {
         final String appID = IntegrationTestUtils.safeUniqueTestName(testInfo);
@@ -302,6 +302,9 @@ private void testReprocessingFromScratchAfterResetWithIntermediateUserTopic(fina
 
         // RE-RUN
         streams.start();
+        TestUtils.waitForCondition(() -> streams.state() == KafkaStreams.State.RUNNING,
+                "KafkaStreams not running in time");
+
         final List<KeyValue<Long, Long>> resultRerun = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
         final List<KeyValue<Long, Long>> resultRerun2 = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC_2_RERUN, 40);
         streams.close();