[BEAM-5197] Fix UnboundedSourceWrapper#testWatermarkEmission#7138
[BEAM-5197] Fix UnboundedSourceWrapper#testWatermarkEmission#7138tweise merged 3 commits intoapache:masterfrom
Conversation
A small wait interval could deadlock on some platforms because the checkpoint lock would be held too often, making it impossible for the source to make progress.
|
I observed that this would fail even on MacOs with a small enough wait interval, e.g. 10 milliseconds. The lock is not held while sleeping but the wait interval is small enough that the source thread may not acquire the lock fast enough before it is acquired again by the test. Would appreciate if someone with Linux could verify this. |
|
|
||
| // Consider that UnboundedSourceWrapper needs to acquire the checkpoint lock below. | ||
| // So wait for enough time for that to happen. | ||
| Thread.sleep(200); |
There was a problem hiding this comment.
Can we use a CountdownLatch or something like that instead of a sleep?
There was a problem hiding this comment.
It is tricky because the processing time is expected to be advanced for the event time timers to progress.
I've refactored the test to wait for the readers to finish; only then the time is advanced and the final trigger executed. That avoids constantly holding the lock.
|
Here is my attempt at it: #7140 |
…estWatermarkEmission
|
Just wondering why this test is needed at all? Doesn't |
|
Good point. I've extended the test to at least check that two watermarks are emitted after the watermark emission is triggered twice. That should make it a good addition to the value test where only the final watermark is checked. |
|
Test failures in latest PreCommit are unrelated to this PR. edit: Also had to fix checkstyle, so tests are running again. |
…er#testWatermarkEmission
| synchronized (checkpointLock) { | ||
| testHarness.setProcessingTime(Instant.now().getMillis()); | ||
| } | ||
| // Need to advance this so that the watermark timers in the source wrapper fire |
There was a problem hiding this comment.
AFAIK the comment still holds true.
| // Need to advance this so that the watermark timers in the source wrapper fire | ||
| // Synchronize is necessary because this can interfere with updating the PriorityQueue | ||
| // of the ProcessingTimeService which is also accessed through UnboundedSourceWrapper. | ||
| synchronized (testHarness.getCheckpointLock()) { |
There was a problem hiding this comment.
This can be moved up to initialization and synchronized won't be needed then.
There was a problem hiding this comment.
Unfortunately this is not possible. The watermark timer is not registered then so this would not to lead to a watermark emission, hence the waiting before for the readers before.
There was a problem hiding this comment.
Now I see that you changed the initial value to Long.MIN_VALUE. What is the significance of changing the processing time to 0 in the next line?
There was a problem hiding this comment.
The idea is to make this test more meaningful by testing emission of multiple watermarks by the source. The initial value before was arbitrary as well, as it was System.currentTimeMillis().
Afterwards we advance to 0, but it could be any value between MIN_VALUE and MAX_VALUE, but it has to be larger than currentTime + autowatermark interval. This emits a watermark from the TestCoutingSource.
Then we advance it to MAX_VALUE to test the emission of the final watermark.
|
ElasticSearchIO test failures. |
|
Run Java PreCommit |
tweise
changed the title
…7138) * Avoid holding checkpoint lock in UnboundedSourceWrapper#testWatermarkEmission * Test emitting multiple watermarks in UnboundedSourceWrapper#testWatermarkEmission
…7138) * Avoid holding checkpoint lock in UnboundedSourceWrapper#testWatermarkEmission * Test emitting multiple watermarks in UnboundedSourceWrapper#testWatermarkEmission
…7138) * Avoid holding checkpoint lock in UnboundedSourceWrapper#testWatermarkEmission * Test emitting multiple watermarks in UnboundedSourceWrapper#testWatermarkEmission
…7138) * Avoid holding checkpoint lock in UnboundedSourceWrapper#testWatermarkEmission * Test emitting multiple watermarks in UnboundedSourceWrapper#testWatermarkEmission
3 participants
A small wait interval could deadlock on some platforms because the checkpoint
lock would be held too often, making it impossible for the source to make
progress.
CC @kennknowles @tweise
Post-Commit Tests Status (on master branch)