From fc1efb720c9c0033077c3c20ee144d0f757e6bcd Mon Sep 17 00:00:00 2001 From: Adam Roberts <aroberts@uk.ibm.com> Date: Fri, 16 Sep 2016 10:20:50 +0100 Subject: [PATCH] [SPARK-17534][TESTS] Increase timeouts for DirectKafkaStreamSuite tests **## What changes were proposed in this pull request?** There are two tests in this suite that are particularly flaky on the following hardware: 2x Intel(R) Xeon(R) CPU E5-2697 v2 2.70GHz and 16 GB of RAM, 1 TB HDD This simple PR increases the timeout times and batch duration so they can reliably pass **## How was this patch tested?** Existing unit tests with the two core box where I was seeing the failures often Author: Adam Roberts <aroberts@uk.ibm.com> Closes #15094 from a-roberts/patch-6. --- .../spark/streaming/kafka010/DirectKafkaStreamSuite.scala | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/DirectKafkaStreamSuite.scala b/external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/DirectKafkaStreamSuite.scala index b1d90b8a82..e04f35eceb 100644 --- a/external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/DirectKafkaStreamSuite.scala +++ b/external/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/DirectKafkaStreamSuite.scala @@ -108,7 +108,7 @@ class DirectKafkaStreamSuite val expectedTotal = (data.values.sum * topics.size) - 2 val kafkaParams = getKafkaParams("auto.offset.reset" -> "earliest") - ssc = new StreamingContext(sparkConf, Milliseconds(200)) + ssc = new StreamingContext(sparkConf, Milliseconds(1000)) val stream = withClue("Error creating direct stream") { KafkaUtils.createDirectStream[String, String]( ssc, @@ -150,7 +150,7 @@ class DirectKafkaStreamSuite allReceived.addAll(Arrays.asList(rdd.map(r => (r.key, r.value)).collect(): _*)) } ssc.start() - eventually(timeout(20000.milliseconds), interval(200.milliseconds)) { + eventually(timeout(100000.milliseconds), interval(1000.milliseconds)) { assert(allReceived.size === expectedTotal, "didn't get expected number of messages, messages:\n" + allReceived.asScala.mkString("\n")) @@ -172,7 +172,7 @@ class DirectKafkaStreamSuite val expectedTotal = (data.values.sum * 2) - 3 val kafkaParams = getKafkaParams("auto.offset.reset" -> "earliest") - ssc = new StreamingContext(sparkConf, Milliseconds(200)) + ssc = new StreamingContext(sparkConf, Milliseconds(1000)) val stream = withClue("Error creating direct stream") { KafkaUtils.createDirectStream[String, String]( ssc, @@ -214,7 +214,7 @@ class DirectKafkaStreamSuite allReceived.addAll(Arrays.asList(rdd.map(r => (r.key, r.value)).collect(): _*)) } ssc.start() - eventually(timeout(20000.milliseconds), interval(200.milliseconds)) { + eventually(timeout(100000.milliseconds), interval(1000.milliseconds)) { assert(allReceived.size === expectedTotal, "didn't get expected number of messages, messages:\n" + allReceived.asScala.mkString("\n")) -- GitLab