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 1 commit
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
7 changes: 6 additions & 1 deletion 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 @@ -45,7 +45,12 @@ class ContinuousSuiteBase extends StreamTest {
case ContinuousScanExec(_, _, r: RateStreamContinuousStream, _) => r
}.get

val deltaMs = numTriggers * 1000 + 300
// Adding 3s in case of slow initialization of partition reader - rows will be committed
// on epoch which they're written.
// Since previous epochs should be committed before to commit the epoch which output rows
// are written, slow initialization of partition reader and tiny trigger interval leads
// output rows to wait long time to be committed.
val deltaMs = numTriggers * 1000 + 3000
Copy link
Member

@dongjoon-hyun dongjoon-hyun Jul 4, 2019

Choose a reason for hiding this comment

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

Thank you for investigating this, @HeartSaVioR .
The original logic looks not safe. Do you think if there is any other better way?

Copy link
Contributor

@gaborgsomogyi gaborgsomogyi Jul 4, 2019

Choose a reason for hiding this comment

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

I've put the test in a loop left it for couple of hours and haven't seen any problem.
On the other hand I agree with @dongjoon-hyun, the original waitForaWhile approach is not safe.

Copy link
Contributor Author

@HeartSaVioR HeartSaVioR Jul 4, 2019

Choose a reason for hiding this comment

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

I'm not sure I can perfectly track when these rows are committed (the safest way), as we are checking this from driver side whereas things are delayed in executor side.

But given the main delay is occurred before first epoch being committed, instead of adding more gaps, we may be able to add 2s (which would be enough for 4 rows to be emitted in rate source) after waiting first epoch to be committed. I'll make a change.

Copy link
Contributor Author

@HeartSaVioR HeartSaVioR Jul 4, 2019

Choose a reason for hiding this comment

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

Never mind. I found we can track the highest committed value, so we can wait for specific value to be committed. Updated the patch.

Copy link
Member

@dongjoon-hyun dongjoon-hyun Jul 5, 2019

Choose a reason for hiding this comment

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

Great!

while (System.currentTimeMillis < reader.creationTime + deltaMs) {
Thread.sleep(reader.creationTime + deltaMs - System.currentTimeMillis)
}
Expand Down