[kafka] branch trunk updated: MINOR: fix race condition in KafkaStreamsTest (#6185)

[gu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

guozhang 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 86995ad  MINOR: fix race condition in KafkaStreamsTest (#6185)
86995ad is described below

commit 86995adbdedb41766eb19d66c2683bc468b295b7
Author: Matthias J. Sax <mj...@apache.org>
AuthorDate: Wed Jan 23 16:55:54 2019 -0800

    MINOR: fix race condition in KafkaStreamsTest (#6185)
    
    Reviewers: John Roesler <jo...@confluent.io>, Guozhang Wang <gu...@confluent.io>
---
 .../java/org/apache/kafka/streams/KafkaStreamsTest.java | 17 ++++-------------
 1 file changed, 4 insertions(+), 13 deletions(-)

diff --git a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
index cc4a9cd..123bd21 100644
--- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java
@@ -84,7 +84,6 @@ import static org.junit.Assert.fail;
 @Category({IntegrationTest.class})
 public class KafkaStreamsTest {
 
-    private static final long TIMEOUT_MS = 30_000L;
     private static final int NUM_BROKERS = 1;
     private static final int NUM_THREADS = 2;
     // We need this to avoid the KafkaConsumer hanging on poll
@@ -158,7 +157,6 @@ public class KafkaStreamsTest {
 
         TestUtils.waitForCondition(
             () -> stateListener.numChanges == 2,
-            TIMEOUT_MS,
             "Streams never started.");
         Assert.assertEquals(KafkaStreams.State.RUNNING, globalStreams.state());
 
@@ -209,7 +207,9 @@ public class KafkaStreamsTest {
 
         globalStreams.close();
 
-        Assert.assertEquals(6, stateListener.numChanges);
+        TestUtils.waitForCondition(
+            () -> stateListener.numChanges == 6,
+            "Streams never closed.");
         Assert.assertEquals(KafkaStreams.State.NOT_RUNNING, globalStreams.state());
     }
 
@@ -226,7 +226,6 @@ public class KafkaStreamsTest {
         streams.close();
         TestUtils.waitForCondition(
             () -> streams.state() == KafkaStreams.State.NOT_RUNNING,
-            TIMEOUT_MS,
             "Streams never stopped.");
 
         // Ensure that any created clients are closed
@@ -254,7 +253,6 @@ public class KafkaStreamsTest {
             streams.start();
             TestUtils.waitForCondition(
                 () -> streams.state() == KafkaStreams.State.RUNNING,
-                TIMEOUT_MS,
                 "Streams never started.");
 
             for (int i = 0; i < NUM_THREADS; i++) {
@@ -262,13 +260,11 @@ public class KafkaStreamsTest {
                 tmpThread.shutdown();
                 TestUtils.waitForCondition(
                     () -> tmpThread.state() == StreamThread.State.DEAD,
-                    TIMEOUT_MS,
                     "Thread never stopped.");
                 threads[i].join();
             }
             TestUtils.waitForCondition(
                 () -> streams.state() == KafkaStreams.State.ERROR,
-                TIMEOUT_MS,
                 "Streams never stopped.");
         } finally {
             streams.close();
@@ -276,7 +272,6 @@ public class KafkaStreamsTest {
 
         TestUtils.waitForCondition(
             () -> streams.state() == KafkaStreams.State.NOT_RUNNING,
-            TIMEOUT_MS,
             "Streams never stopped.");
 
         final java.lang.reflect.Field globalThreadField = streams.getClass().getDeclaredField("globalStreamThread");
@@ -295,7 +290,6 @@ public class KafkaStreamsTest {
             streams.start();
             TestUtils.waitForCondition(
                 () -> streams.state() == KafkaStreams.State.RUNNING,
-                TIMEOUT_MS,
                 "Streams never started.");
             final java.lang.reflect.Field globalThreadField = streams.getClass().getDeclaredField("globalStreamThread");
             globalThreadField.setAccessible(true);
@@ -303,7 +297,6 @@ public class KafkaStreamsTest {
             globalStreamThread.shutdown();
             TestUtils.waitForCondition(
                 () -> globalStreamThread.state() == GlobalStreamThread.State.DEAD,
-                TIMEOUT_MS,
                 "Thread never stopped.");
             globalStreamThread.join();
             assertEquals(streams.state(), KafkaStreams.State.ERROR);
@@ -555,7 +548,6 @@ public class KafkaStreamsTest {
         globalStreams.start();
         TestUtils.waitForCondition(
             () -> globalStreams.state() == KafkaStreams.State.RUNNING,
-            TIMEOUT_MS,
             "Streams never started.");
 
         try {
@@ -613,7 +605,7 @@ public class KafkaStreamsTest {
         th.start();
 
         try {
-            th.join(TIMEOUT_MS);
+            th.join(30_000L);
             assertFalse(th.isAlive());
         } finally {
             streams.close();
@@ -763,7 +755,6 @@ public class KafkaStreamsTest {
         final File taskDir = new File(appDir, "0_0");
         TestUtils.waitForCondition(
             () -> !oldTaskDir.exists() && taskDir.exists(),
-            TIMEOUT_MS,
             "cleanup has not successfully run");
         assertTrue(taskDir.exists());
     }