GitHub user tdas opened a pull request:
https://github.com/apache/spark/pull/22182
[SPARK-25184][SS] Fixed race condition in StreamExecution that caused flaky
test in FlatMapGroupsWithState
## What changes were proposed in this pull request?
The race condition that caused test failure is between 2 threads.
- The MicrobatchExecution thread that processes inputs to produce answers
and then generates progress events.
- The test thread that generates some input data, checked the answer and
then verified the query generated progress event.
The synchronization structure between these threads is as follows
1. MicrobatchExecution thread, in every batch, does the following in order.
a. Processes batch input to generate answer.
b. Signals `awaitProgressLockCondition` to wake up threads waiting for
progress using `awaitOffset`
c. Generates progress event
2. Test execution thread
a. Calls `awaitOffset` to wait for progress, which waits on
`awaitProgressLockCondition`.
b. As soon as `awaitProgressLockCondition` is signaled, it would move on
the in the test to check answer.
c. Finally, it would verify the last generated progress event.
What can happen is the following sequence of events: 2a -> 1a -> 1b -> 2b
-> 2c -> 1c.
In other words, the progress event may be generated after the test tries to
verify it.
The solution has two steps.
1. Signal the waiting thread after the progress event has been generated,
that is, after `finishTrigger()`.
2. Increase the timeout of `awaitProgressLockCondition.await(100 ms)` to a
large value.
This latter is to ensure that test thread for keeps waiting on
`awaitProgressLockCondition`until the MicroBatchExecution thread explicitly
signals it. With the existing small timeout of 100ms the following sequence can
occur.
- MicroBatchExecution thread updates committed offsets
- Test thread waiting on `awaitProgressLockCondition` accidentally times
out after 100 ms, finds that the committed offsets have been updated, therefore
returns from `awaitOffset` and moves on to the progress event tests.
- MicroBatchExecution thread then generates progress event and signals.
But the test thread has already attempted to verify the event and failed.
By increasing the timeout to large (e.g., `streamingTimeoutMs = 60
seconds`, similar to `awaitInitialization`), this above type of race condition
is also avoided.
## How was this patch tested?
Ran locally many times.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/tdas/spark SPARK-25184
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/22182.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #22182
commit 319990ff60ad7b6fad6fd0cea5cada0b22e3f3c9
Author: Tathagata Das
Date: 2018-08-22T04:44:59Z
[SC-12136][SS][HOTFIX] Fixed race condition in StreamExecution that caused
flaky test in FlatMapGroupsWithState
The race condition that caused test failure is between 2 threads.
- The MicrobatchExecution thread that processes inputs to produce answers
and then generates progress events.
- The test thread that generates some input data, checked the answer and
then verified the query generated progress event.
The synchronization structure between these threads is as follows
1. MicrobatchExecution thread, in every batch, does the following in order.
a. Processes batch input to generate answer.
b. Signals `awaitProgressLockCondition` to wake up threads waiting for
progress using `awaitOffset`
c. Generates progress event
2. Test execution thread
a. Calls `awaitOffset` to wait for progress, which waits on
`awaitProgressLockCondition`.
b. As soon as `awaitProgressLockCondition` is signaled, it would move on
the in the test to check answer.
c. Finally, it would verify the last generated progress event.
What can happen is the following sequence of events: 2a -> 1a -> 1b -> 2b
-> 2c -> 1c.
In other words, the progress event may be generated after the test tries to
verify it.
The solution has two steps.
1. Signal the waiting thread after the progress event has been generated,
that is, after `finishTrigger()`.
2. Increase the timeout of `awaitProgressLockCondition.await(100 ms)` to a
large value.
This latter is to ensure that test thread for keeps waiting on
`awaitProgressLockCondition`until the MicroBatchExecution thread explicitly
signals it. With the existing small timeout of 100ms the following sequence can
occur.
- MicroBatchExecution thread updates committed offsets
- Test thread waiting on `awaitProgressLockCondition` ac
