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

Show file tree
Hide file tree
Changes from 4 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
16 changes: 15 additions & 1 deletion ...cala/org/apache/spark/sql/execution/streaming/continuous/ContinuousRateStreamSource.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.spark.sql.execution.streaming.continuous

import java.util.concurrent.atomic.AtomicLong

import org.json4s.DefaultFormats
import org.json4s.jackson.Serialization

Expand All @@ -36,6 +38,9 @@ class RateStreamContinuousStream(rowsPerSecond: Long, numPartitions: Int) extend

val perPartitionRate = rowsPerSecond.toDouble / numPartitions.toDouble

val highestCommittedValue = new AtomicLong(Long.MinValue)
val firstCommittedTime = new AtomicLong(Long.MinValue)

@dongjoon-hyun dongjoon-hyun Jul 5, 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.

So, do we need to add these variable only for testing? Is there any other valuable information about these?

@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.

Actually yes, and creationTime is also only used from tests. Rate stream is designed for test and evaluation purpose, so it seems OK to modify rate stream to support tests.

Btw, I'll adjust the scope to private[sql], as it doesn't need to be exposed outside of Spark.


override def mergeOffsets(offsets: Array[PartitionOffset]): Offset = {
assert(offsets.length == numPartitions)
val tuples = offsets.map {
Expand Down Expand Up @@ -82,7 +87,16 @@ class RateStreamContinuousStream(rowsPerSecond: Long, numPartitions: Int) extend
RateStreamContinuousReaderFactory
}

override def commit(end: Offset): Unit = {}
override def commit(end: Offset): Unit = {

@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 change shouldn't bring noticeable perf hit, as it is only called per epoch which interval would be at least hundreds of milliseconds.

end.asInstanceOf[RateStreamOffset].partitionToValueAndRunTimeMs.foreach {
case (_, ValueRunTimeMsPair(value, _)) =>
if (highestCommittedValue.get() < value) {

@srowen srowen Jul 8, 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'm not sure if the 'atomic' part is essential here, but if it is, I think you have a race condition here. You'd want to use updateAndGet or something to make sure the check and update are atomic.

@HeartSaVioR HeartSaVioR Jul 8, 2019 •
edited
Loading

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.

Thanks for pointing out! Only one writer and one reader will run concurrently. I'm revisiting the change, and it looks like just over-engineering. volatile would just work. If reader is reading the old value they just need to wait a bit more, so not strictly need to have atomicity on update. I'll make a change.

highestCommittedValue.set(value)
}
}
firstCommittedTime.compareAndSet(Long.MinValue, System.currentTimeMillis())

@HeartSaVioR HeartSaVioR Jul 8, 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.

Similar here: one writer and one reader, and we have alternative logic (in waitForRateSourceTriggers) when reader reads old value so atomicity is not strictly needed.

}

override def stop(): Unit = {}

private def createInitialOffset(numPartitions: Int, creationTimeMs: Long) = {
Expand Down
49 changes: 36 additions & 13 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 @@ -38,18 +38,42 @@ class ContinuousSuiteBase extends StreamTest {
sparkConf.set("spark.sql.testkey", "true")))

protected def waitForRateSourceTriggers(query: StreamExecution, numTriggers: Int): Unit = {
query match {
findRateStreamContinuousStream(query).foreach { reader =>
val deltaMs = numTriggers * 1000 + 300
val firstCommittedTime = reader.firstCommittedTime.longValue()

@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 change is based on the observation: while reader.creationTime doesn't guarantee partition readers are initialized. Instead, reader.firstCommittedTime guarantees partition readers are initialized before, so we can capture and use this.
(This is a best-effort on driver side, not fastest approach, of course.)

while (System.currentTimeMillis < firstCommittedTime + deltaMs) {
Thread.sleep(firstCommittedTime + deltaMs - System.currentTimeMillis)
}
}
}

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 = {
findRateStreamContinuousStream(query).foreach { reader =>
val startTime = System.currentTimeMillis()
val maxWait = startTime + maxWaitTimeMs
while (System.currentTimeMillis() < maxWait &&
reader.highestCommittedValue.get() < desiredValue) {
Thread.sleep(100)
}
if (System.currentTimeMillis() > maxWait) {
logWarning(s"Couldn't reach desired value in $maxWaitTimeMs milliseconds!" +
s"Current highest committed value is ${reader.highestCommittedValue}")
}
}
}

private def findRateStreamContinuousStream(
query: StreamExecution): Option[RateStreamContinuousStream] = 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 {
s.lastExecution.executedPlan.collectFirst {
case ContinuousScanExec(_, _, r: RateStreamContinuousStream, _) => r
}.get

val deltaMs = numTriggers * 1000 + 300
while (System.currentTimeMillis < reader.creationTime + deltaMs) {
Thread.sleep(reader.creationTime + deltaMs - System.currentTimeMillis)
}
}

case _ => None
}

// A continuous trigger that will only fire the initial time for the duration of a test.
Expand Down Expand Up @@ -218,8 +242,7 @@ class ContinuousSuite extends ContinuousSuiteBase {
.start()
val continuousExecution =
query.asInstanceOf[StreamingQueryWrapper].streamingQuery.asInstanceOf[ContinuousExecution]
continuousExecution.awaitEpoch(0)
waitForRateSourceTriggers(continuousExecution, 2)
waitForRateSourceCommittedValue(continuousExecution, 3, 20 * 1000)
query.stop()

val results = spark.read.table("noharness").collect()
Expand All @@ -241,7 +264,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 +282,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