You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by yh...@apache.org on 2016/03/10 23:38:23 UTC
spark git commit: [SQL][TEST] Increased timeouts to reduce flakiness
in ContinuousQueryManagerSuite
Repository: spark
Updated Branches:
refs/heads/master 747d2f538 -> 3d2b6f56e
[SQL][TEST] Increased timeouts to reduce flakiness in ContinuousQueryManagerSuite
## What changes were proposed in this pull request?
ContinuousQueryManager is sometimes flaky on Jenkins. I could not reproduce it on my machine, so I guess it about the waiting times which causes problems if Jenkins is loaded. I have increased the wait time in the hope that it will be less flaky.
## How was this patch tested?
I reran the unit test many times on a loop in my machine. I am going to run it a few time in Jenkins, that's the real test.
Author: Tathagata Das <ta...@gmail.com>
Closes #11638 from tdas/cqm-flaky-test.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/3d2b6f56
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/3d2b6f56
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/3d2b6f56
Branch: refs/heads/master
Commit: 3d2b6f56e38ce867ae8819752fd693adab9a8cc9
Parents: 747d2f5
Author: Tathagata Das <ta...@gmail.com>
Authored: Thu Mar 10 14:38:19 2016 -0800
Committer: Yin Huai <yh...@databricks.com>
Committed: Thu Mar 10 14:38:19 2016 -0800
----------------------------------------------------------------------
.../streaming/ContinuousQueryManagerSuite.scala | 30 ++++++++++----------
1 file changed, 15 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/3d2b6f56/sql/core/src/test/scala/org/apache/spark/sql/streaming/ContinuousQueryManagerSuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/ContinuousQueryManagerSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/ContinuousQueryManagerSuite.scala
index 35bb9fd..45e824a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/streaming/ContinuousQueryManagerSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/ContinuousQueryManagerSuite.scala
@@ -148,9 +148,9 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
// awaitAnyTermination should be blocking or non-blocking depending on timeout values
testAwaitAnyTermination(
ExpectBlocked,
- awaitTimeout = 2 seconds,
+ awaitTimeout = 4 seconds,
expectedReturnedValue = false,
- testBehaviorFor = 1 second)
+ testBehaviorFor = 2 seconds)
testAwaitAnyTermination(
ExpectNotBlocked,
@@ -162,20 +162,20 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
val q1 = stopRandomQueryAsync(stopAfter = 100 milliseconds, withError = false)
testAwaitAnyTermination(
ExpectNotBlocked,
- awaitTimeout = 1 second,
+ awaitTimeout = 2 seconds,
expectedReturnedValue = true,
- testBehaviorFor = 2 seconds)
+ testBehaviorFor = 4 seconds)
require(!q1.isActive) // should be inactive by the time the prev awaitAnyTerm returned
// All subsequent calls to awaitAnyTermination should be non-blocking even if timeout is high
testAwaitAnyTermination(
- ExpectNotBlocked, awaitTimeout = 2 seconds, expectedReturnedValue = true)
+ ExpectNotBlocked, awaitTimeout = 4 seconds, expectedReturnedValue = true)
// Resetting termination should make awaitAnyTermination() blocking again
sqlContext.streams.resetTerminated()
testAwaitAnyTermination(
ExpectBlocked,
- awaitTimeout = 2 seconds,
+ awaitTimeout = 4 seconds,
expectedReturnedValue = false,
testBehaviorFor = 1 second)
@@ -184,31 +184,31 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
val q2 = stopRandomQueryAsync(100 milliseconds, withError = true)
testAwaitAnyTermination(
ExpectException[SparkException],
- awaitTimeout = 1 second,
+ awaitTimeout = 1 seconds,
testBehaviorFor = 2 seconds)
require(!q2.isActive) // should be inactive by the time the prev awaitAnyTerm returned
// All subsequent calls to awaitAnyTermination should throw the exception
testAwaitAnyTermination(
ExpectException[SparkException],
- awaitTimeout = 1 second,
- testBehaviorFor = 2 seconds)
+ awaitTimeout = 2 seconds,
+ testBehaviorFor = 4 seconds)
// Terminate a query asynchronously outside the timeout, awaitAnyTerm should be blocked
sqlContext.streams.resetTerminated()
- val q3 = stopRandomQueryAsync(1 second, withError = true)
+ val q3 = stopRandomQueryAsync(2 seconds, withError = true)
testAwaitAnyTermination(
ExpectNotBlocked,
awaitTimeout = 100 milliseconds,
expectedReturnedValue = false,
- testBehaviorFor = 2 seconds)
+ testBehaviorFor = 4 seconds)
// After that query is stopped, awaitAnyTerm should throw exception
eventually(Timeout(streamingTimeout)) { require(!q3.isActive) } // wait for query to stop
testAwaitAnyTermination(
ExpectException[SparkException],
awaitTimeout = 100 milliseconds,
- testBehaviorFor = 2 seconds)
+ testBehaviorFor = 4 seconds)
// Terminate multiple queries, one with failure and see whether awaitAnyTermination throws
@@ -217,12 +217,12 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
val q4 = stopRandomQueryAsync(10 milliseconds, withError = false)
testAwaitAnyTermination(
- ExpectNotBlocked, awaitTimeout = 1 second, expectedReturnedValue = true)
+ ExpectNotBlocked, awaitTimeout = 2 seconds, expectedReturnedValue = true)
require(!q4.isActive)
val q5 = stopRandomQueryAsync(10 milliseconds, withError = true)
eventually(Timeout(streamingTimeout)) { require(!q5.isActive) }
// After q5 terminates with exception, awaitAnyTerm should start throwing exception
- testAwaitAnyTermination(ExpectException[SparkException], awaitTimeout = 100 milliseconds)
+ testAwaitAnyTermination(ExpectException[SparkException], awaitTimeout = 2 seconds)
}
}
@@ -260,7 +260,7 @@ class ContinuousQueryManagerSuite extends StreamTest with SharedSQLContext with
expectedBehavior: ExpectedBehavior,
expectedReturnedValue: Boolean = false,
awaitTimeout: Span = null,
- testBehaviorFor: Span = 2 seconds
+ testBehaviorFor: Span = 4 seconds
): Unit = {
def awaitTermFunc(): Unit = {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org