MINOR: Follow-up from KAFKA-3842 to fix tempDir

bbejeck · guozhangwang · commit 96ae0a5929bc · 2016-06-29T15:17:04.000-07:00
…p directories, waitForCondition Author: bbejeck <bbejeck@gmail.com> Reviewers: Guozhang Wang, Ismael Juma Closes apache#1554 from bbejeck/follow_up_for_KAFKA-3842
diff --git a/clients/src/test/java/org/apache/kafka/test/TestUtils.java b/clients/src/test/java/org/apache/kafka/test/TestUtils.java
@@ -123,20 +123,17 @@ public static File tempFile() throws IOException {
      *
      * @param prefix The prefix of the temporary directory, if null using "kafka-" as default prefix
      */
-    public static File tempDirectory(String prefix) throws IOException {
+    public static File tempDirectory(String prefix) {
         return tempDirectory(null, prefix);
     }
 
     /**
-     * Create a temporary directory named "test" under /temp
+     * Create a temporary relative directory in the default temporary-file directory with a
+     * prefix of "kafka-"
      * @return  the temporary directory just created.
      */
-    public static File tempDir() {
-        try {
-            return tempDirectory(new File("/tmp").toPath(), "test");
-        } catch (IOException ex) {
-            throw new RuntimeException("Failed to create a temp dir", ex);
-        }
+    public static File tempDirectory() {
+        return tempDirectory(null);
     }
 
     /**
@@ -145,10 +142,15 @@ public static File tempDir() {
      * @param parent The parent folder path name, if null using the default temporary-file directory
      * @param prefix The prefix of the temporary directory, if null using "kafka-" as default prefix
      */
-    public static File tempDirectory(Path parent, String prefix) throws IOException {
-        final File file = parent == null ?
-                Files.createTempDirectory(prefix == null ? "kafka-" : prefix).toFile() :
-                Files.createTempDirectory(parent, prefix == null ? "kafka-" : prefix).toFile();
+    public static File tempDirectory(Path parent, String prefix) {
+        final File file;
+        prefix = prefix == null ? "kafka-" : prefix;
+        try {
+            file = parent == null ?
+                    Files.createTempDirectory(prefix).toFile() : Files.createTempDirectory(parent, prefix).toFile();
+        } catch (IOException ex) {
+            throw new RuntimeException("Failed to create a temp dir", ex);
+        }
         file.deleteOnExit();
 
         Runtime.getRuntime().addShutdownHook(new Thread() {
@@ -225,21 +227,29 @@ public static Properties consumerConfig(final String bootstrapServers, Class key
     }
 
     /**
-     *  uses default value of 30 seconds for timeout
+     *  uses default value of 15 seconds for timeout
      */
-    public static void waitForCondition(TestCondition testCondition) throws InterruptedException {
-        waitForCondition(testCondition, 30000);
+    public static void waitForCondition(TestCondition testCondition, String conditionDetails) throws InterruptedException {
+        waitForCondition(testCondition, 15000, conditionDetails);
     }
 
     /**
-     *  Used to wait for specific conditions/state to be me during a test
-     *  this is meant to be a replacement for using Thread.sleep
+     * Wait for condition to be met for at most {@code maxWaitMs} and throw assertion failure otherwise.
+     * This should be used instead of {@code Thread.sleep} whenever possible as it allows a longer timeout to be used
+     * without unnecessarily increasing test time (as the condition is checked frequently). The longer timeout is needed to
+     * avoid transient failures due to slow or overloaded machines.
      */
-    public static void waitForCondition(TestCondition testCondition, long maxTimeMillis) throws InterruptedException {
+    public static void waitForCondition(TestCondition testCondition, long maxWaitMs, String conditionDetails) throws InterruptedException {
         long startTime = System.currentTimeMillis();
 
-        while (!testCondition.conditionMet() && ((System.currentTimeMillis() - startTime) < maxTimeMillis)) {
-            Thread.sleep(Math.min(maxTimeMillis, 100L));
+
+        while (!testCondition.conditionMet() && ((System.currentTimeMillis() - startTime) < maxWaitMs)) {
+            Thread.sleep(Math.min(maxWaitMs, 100L));
+        }
+
+        if (!testCondition.conditionMet()) {
+            conditionDetails = conditionDetails != null ? conditionDetails : "";
+            throw new AssertionError("Condition not met within timeout " + maxWaitMs + ". " + conditionDetails);
         }
     }
 
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
@@ -123,7 +123,7 @@ public void shouldCompactTopicsForStateChangelogs() throws Exception {
         streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, CLUSTER.zKConnectString());
         streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
         streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
-        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDir().getPath());
+        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
         streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
         KStreamBuilder builder = new KStreamBuilder();
 
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/JoinIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/JoinIntegrationTest.java
@@ -149,7 +149,7 @@ public void shouldCountClicksPerRegion() throws Exception {
         // with automatically) we don't need to set this anymore and can update `purgeLocalStreamsState`
         // accordingly.
         streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG,
-                                 TestUtils.tempDir().getPath());
+                                 TestUtils.tempDirectory().getPath());
 
         // Remove any state from previous test runs
         IntegrationTestUtils.purgeLocalStreamsState(streamsConfiguration);
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KGroupedStreamIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KGroupedStreamIntegrationTest.java
@@ -80,7 +80,7 @@ public void before() {
             .put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
         streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, CLUSTER.zKConnectString());
         streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDir().getPath());
+        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
 
         KeyValueMapper<Integer, String, String>
             mapper =
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/KStreamRepartitionJoinTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/KStreamRepartitionJoinTest.java
@@ -83,7 +83,7 @@ public void before() {
             .put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
         streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, CLUSTER.zKConnectString());
         streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
-        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDir().getPath());
+        streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
         streamsConfiguration.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3);
 
 
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
@@ -86,6 +86,7 @@ public class RegexSourceIntegrationTest {
     private static final String DEFAULT_OUTPUT_TOPIC = "outputTopic";
     private static final String STRING_SERDE_CLASSNAME = Serdes.String().getClass().getName();
     private Properties streamsConfiguration;
+    private static final String STREAM_TASKS_NOT_UPDATED = "Stream tasks not updated";
 
 
     @BeforeClass
@@ -146,11 +147,12 @@ public void testRegexMatchesTopicsAWhenCreated() throws Exception {
         streamThreads[0] = testStreamThread;
         streams.start();
 
-        TestUtils.waitForCondition(tasksUpdated);
+        TestUtils.waitForCondition(tasksUpdated, STREAM_TASKS_NOT_UPDATED);
+        testStreamThread.streamTaskUpdated = false;
 
         CLUSTER.createTopic("TEST-TOPIC-2");
 
-        TestUtils.waitForCondition(tasksUpdated);
+        TestUtils.waitForCondition(tasksUpdated, STREAM_TASKS_NOT_UPDATED);
 
         streams.close();
 
@@ -193,11 +195,13 @@ public void testRegexMatchesTopicsAWhenDeleted() throws Exception {
 
         streams.start();
 
-        TestUtils.waitForCondition(tasksUpdated);
+        TestUtils.waitForCondition(tasksUpdated, STREAM_TASKS_NOT_UPDATED);
+        //reset
+        testStreamThread.streamTaskUpdated = false;
 
         CLUSTER.deleteTopic("TEST-TOPIC-A");
 
-        TestUtils.waitForCondition(tasksUpdated);
+        TestUtils.waitForCondition(tasksUpdated, STREAM_TASKS_NOT_UPDATED);
 
         streams.close();
 
@@ -331,12 +335,7 @@ private TestCondition createTasksUpdatedCondition(final TestStreamThread testStr
         return new TestCondition() {
             @Override
             public boolean conditionMet() {
-                if (testStreamThread.streamTaskUpdated) {
-                    testStreamThread.streamTaskUpdated = false;
-                    return true;
-                } else {
-                    return false;
-                }
+                return testStreamThread.streamTaskUpdated;
             }
         };
     }
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
@@ -200,13 +200,9 @@ public boolean conditionMet() {
             }
         };
 
-        TestUtils.waitForCondition(valuesRead, waitTime);
+        String conditionDetails = "Did not receive " + expectedNumRecords + " number of records";
 
-        if (accumData.size() < expectedNumRecords) {
-            throw new AssertionError("Expected " + expectedNumRecords +
-                    " but received only " + accumData.size() +
-                    " records before timeout " + waitTime + " ms");
-        }
+        TestUtils.waitForCondition(valuesRead, waitTime, conditionDetails);
 
         return accumData;
     }
@@ -243,16 +239,11 @@ public boolean conditionMet() {
             }
         };
 
-        TestUtils.waitForCondition(valuesRead, waitTime);
+        String conditionDetails = "Did not receive " + expectedNumRecords + " number of records";
 
-        if (accumData.size() < expectedNumRecords) {
-            throw new AssertionError("Expected " + expectedNumRecords +
-                    " but received only " + accumData.size() +
-                    " records before timeout " + waitTime + " ms");
-        }
+        TestUtils.waitForCondition(valuesRead, waitTime, conditionDetails);
 
         return accumData;
-
     }
 
 }
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java
@@ -66,7 +66,7 @@ public class ProcessorTopologyTest {
     @Before
     public void setup() {
         // Create a new directory in which we'll put all of the state for this test, enabling running tests in parallel ...
-        File localState = TestUtils.tempDir();
+        File localState = TestUtils.tempDirectory();
         Properties props = new Properties();
         props.setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "processor-topology-test");
         props.setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9091");
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java b/streams/src/test/java/org/apache/kafka/streams/state/KeyValueStoreTestDriver.java
@@ -211,7 +211,7 @@ public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySeria
                 send(record, keySerializer, valueSerializer);
             }
         };
-        this.stateDir = TestUtils.tempDir();
+        this.stateDir = TestUtils.tempDirectory();
         this.stateDir.mkdirs();
 
         Properties props = new Properties();

Original file line number	Diff line number	Diff line change
`@@ -200,13 +200,9 @@ public boolean conditionMet() {`
`200`	`200`	`}`
`201`	`201`	`};`
`202`	`202`
`203`		`- TestUtils.waitForCondition(valuesRead, waitTime);`
	`203`	`+ String conditionDetails = "Did not receive " + expectedNumRecords + " number of records";`
`204`	`204`
`205`		`- if (accumData.size() < expectedNumRecords) {`
`206`		`- throw new AssertionError("Expected " + expectedNumRecords +`
`207`		`- " but received only " + accumData.size() +`
`208`		`- " records before timeout " + waitTime + " ms");`
`209`		`- }`
	`205`	`+ TestUtils.waitForCondition(valuesRead, waitTime, conditionDetails);`
`210`	`206`
`211`	`207`	`return accumData;`
`212`	`208`	`}`
`@@ -243,16 +239,11 @@ public boolean conditionMet() {`
`243`	`239`	`}`
`244`	`240`	`};`
`245`	`241`
`246`		`- TestUtils.waitForCondition(valuesRead, waitTime);`
	`242`	`+ String conditionDetails = "Did not receive " + expectedNumRecords + " number of records";`
`247`	`243`
`248`		`- if (accumData.size() < expectedNumRecords) {`
`249`		`- throw new AssertionError("Expected " + expectedNumRecords +`
`250`		`- " but received only " + accumData.size() +`
`251`		`- " records before timeout " + waitTime + " ms");`
`252`		`- }`
	`244`	`+ TestUtils.waitForCondition(valuesRead, waitTime, conditionDetails);`
`253`	`245`
`254`	`246`	`return accumData;`
`255`		`-`
`256`	`247`	`}`
`257`	`248`
`258`	`249`	`}`
Original file line number	Diff line number	Diff line change
`@@ -211,7 +211,7 @@ public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySeria`
`211`	`211`	`send(record, keySerializer, valueSerializer);`
`212`	`212`	`}`
`213`	`213`	`};`
`214`		`- this.stateDir = TestUtils.tempDir();`
	`214`	`+ this.stateDir = TestUtils.tempDirectory();`
`215`	`215`	`this.stateDir.mkdirs();`
`216`	`216`
`217`	`217`	`Properties props = new Properties();`