Skip to content

[SPARK-28247][SS] Fix flaky test "query without test harness" on ContinuousSuite #25048

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HeartSaVioR wants to merge 9 commits into from
Closed

[SPARK-28247][SS] Fix flaky test "query without test harness" on ContinuousSuite #25048

Changes from 7 commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
299f629
[SPARK-28247][SS] Fix flaky test on ContinuousSuite "query without te…
HeartSaVioR Jul 4, 2019
97b5de6
Apply safer approach
HeartSaVioR Jul 4, 2019
2cadef3
Minor refine
HeartSaVioR Jul 4, 2019
c006be8
Change the approach to determine start time of rate stream to when fi…
HeartSaVioR Jul 5, 2019
8e309ac
Fix a bug when first epoch is not reported to rate source yet, adjust…
HeartSaVioR Jul 5, 2019
d3fbd82
add detailed message on verification failure
HeartSaVioR Jul 5, 2019
faa41b5
Remove unnecessary fields via reading from ContinuousExecution instead
HeartSaVioR Jul 8, 2019
aa932bc
Fix indent
HeartSaVioR Jul 10, 2019
6706a65
Reflect review comments
HeartSaVioR Jul 10, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
53 changes: 42 additions & 11 deletions sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,14 +40,43 @@ class ContinuousSuiteBase extends StreamTest {
protected def waitForRateSourceTriggers(query: StreamExecution, numTriggers: Int): Unit = {
query match {
case s: ContinuousExecution =>
assert(numTriggers >= 2, "must wait for at least 2 triggers to ensure query is initialized")
val reader = s.lastExecution.executedPlan.collectFirst {
case ContinuousScanExec(_, _, r: RateStreamContinuousStream, _) => r
}.get
s.awaitEpoch(0)

// This is called after waiting first epoch to be committed, so we can just treat
// it as partition readers for rate source are already initialized.
val firstCommittedTime = System.currentTimeMillis()
val deltaMs = numTriggers * 1000 + 300
while (System.currentTimeMillis < reader.creationTime + deltaMs) {
Thread.sleep(reader.creationTime + deltaMs - System.currentTimeMillis)
while (System.currentTimeMillis < firstCommittedTime + deltaMs) {
Thread.sleep(firstCommittedTime + deltaMs - System.currentTimeMillis)

@srowen srowen Jul 10, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this could cause an exception if the current time advanced between the check and the sleep, as a negative argument will cause an exception. Maybe:

var toWaitMS = firstCommittedTime + deltaMs - System.currentTimeMillis
while (toWaitMS > 0) {
  Thread.sleep(toWaitMS)
  toWaitMS = firstCommittedTime + deltaMs - System.currentTimeMillis
}

@srowen srowen Jul 10, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

(I guess technically you should use System.nanoTime here but it probably won't matter)

@HeartSaVioR HeartSaVioR Jul 10, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You're right we're calling System.currentTimeMillis twice which might not be the same, so possible to be negative. Nice finding.

@HeartSaVioR HeartSaVioR Jul 10, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed.

}
}
}

protected def waitForRateSourceCommittedValue(

@HeartSaVioR HeartSaVioR Jul 5, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is safest approach to expect some rows to produce outputs. We still need to have max time to wait, since it may block infinitely in case of bugs.

query: StreamExecution,
desiredValue: Long,
maxWaitTimeMs: Long): Unit = {
def readHighestCommittedValue(c: ContinuousExecution): Option[Long] = {
c.committedOffsets.lastOption.map { case (_, offset) =>
offset match {
case o: RateStreamOffset =>
o.partitionToValueAndRunTimeMs.map {
case (_, ValueRunTimeMsPair(value, _)) => value
}.max
}
}
}

query match {
case c: ContinuousExecution =>

@srowen srowen Jul 10, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just make the method take a ContinuousExecution if that's all it can meaningfully handle. (Cast in the caller if needed)

@HeartSaVioR HeartSaVioR Jul 10, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It might make callers a bit more verbose than handling it from here, but not a big deal and better in perspective of avoiding no-op when misused. I'll make a change.

@HeartSaVioR HeartSaVioR Jul 10, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Addressed.

val maxWait = System.currentTimeMillis() + maxWaitTimeMs
while (System.currentTimeMillis() < maxWait &&
readHighestCommittedValue(c).getOrElse(Long.MinValue) < desiredValue) {
Thread.sleep(100)
}
if (System.currentTimeMillis() > maxWait) {
logWarning(s"Couldn't reach desired value in $maxWaitTimeMs milliseconds!" +
s"Current highest committed value is ${readHighestCommittedValue(c)}")
}
}
}
Expand Down Expand Up @@ -216,14 +245,16 @@ class ContinuousSuite extends ContinuousSuiteBase {
.queryName("noharness")
.trigger(Trigger.Continuous(100))
.start()

val expected = Set(0, 1, 2, 3)
val continuousExecution =
query.asInstanceOf[StreamingQueryWrapper].streamingQuery.asInstanceOf[ContinuousExecution]
continuousExecution.awaitEpoch(0)
waitForRateSourceTriggers(continuousExecution, 2)
waitForRateSourceCommittedValue(continuousExecution, expected.max, 20 * 1000)
query.stop()

val results = spark.read.table("noharness").collect()
assert(Set(0, 1, 2, 3).map(Row(_)).subsetOf(results.toSet))
assert(expected.map(Row(_)).subsetOf(results.toSet),
s"Result set ${results.toSet} are not a superset of $expected!")

@gaborgsomogyi gaborgsomogyi Jul 10, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: indent

}
}

Expand All @@ -241,7 +272,7 @@ class ContinuousStressSuite extends ContinuousSuiteBase {
testStream(df)(
StartStream(longContinuousTrigger),
AwaitEpoch(0),
Execute(waitForRateSourceTriggers(_, 10)),
Execute(waitForRateSourceTriggers(_, 5)),

@HeartSaVioR HeartSaVioR Jul 5, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The change saves couple of seconds in my machine.

IncrementEpoch(),
StopStream,
CheckAnswerRowsContains(scala.Range(0, 2500).map(Row(_)))
Expand All @@ -259,7 +290,7 @@ class ContinuousStressSuite extends ContinuousSuiteBase {
testStream(df)(
StartStream(Trigger.Continuous(2012)),
AwaitEpoch(0),
Execute(waitForRateSourceTriggers(_, 10)),
Execute(waitForRateSourceTriggers(_, 5)),

@HeartSaVioR HeartSaVioR Jul 5, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ditto.

IncrementEpoch(),
StopStream,
CheckAnswerRowsContains(scala.Range(0, 2500).map(Row(_))))
Expand Down