AsyncTwoPhaseIndexerTests race condition fixed Backport#37830
The unlucky timing can cause this test to fail when the indexing is triggered from maybeTriggerAsyncJob. As this is asynchronous, in can finish quicker then the test stepping over to next assertion
The introduced barrier solves the problem
closes
Backport
@@ -20,9 +20,11 @@ import org.elasticsearch.test.ESTestCase;
import java.io.IOException;
import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -34,11 +36,14 @@ public class AsyncTwoPhaseIndexerTests extends ESTestCase {
private class MockIndexer extends AsyncTwoPhaseIndexer<Integer, MockJobStats> {
+ private final CountDownLatch latch;
// test the execution order
private int step;
- protected MockIndexer(Executor executor, AtomicReference<IndexerState> initialState, Integer initialPosition) {
+ protected MockIndexer(Executor executor, AtomicReference<IndexerState> initialState, Integer initialPosition,
+ CountDownLatch latch) {
super(executor, initialState, initialPosition, new MockJobStats());
+ this.latch = latch;
}
@Override
@@ -48,11 +53,20 @@ public class AsyncTwoPhaseIndexerTests extends ESTestCase {
@Override
protected IterationResult<Integer> doProcess(SearchResponse searchResponse) {
+ awaitForLatch();
assertThat(step, equalTo(3));
++step;
return new IterationResult<Integer>(Collections.emptyList(), 3, true);
}
+ private void awaitForLatch() {
+ try {
+ latch.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
protected SearchRequest buildSearchRequest() {
assertThat(step, equalTo(1));
@@ -195,12 +209,14 @@ public class AsyncTwoPhaseIndexerTests extends ESTestCase {
final ExecutorService executor = Executors.newFixedThreadPool(1);
isFinished.set(false);
try {
-
- MockIndexer indexer = new MockIndexer(executor, state, 2);
+ CountDownLatch countDownLatch = new CountDownLatch(1);
+ MockIndexer indexer = new MockIndexer(executor, state, 2, countDownLatch);
indexer.start();
assertThat(indexer.getState(), equalTo(IndexerState.STARTED));
assertTrue(indexer.maybeTriggerAsyncJob(System.currentTimeMillis()));
assertThat(indexer.getState(), equalTo(IndexerState.INDEXING));
+ countDownLatch.countDown();
+
assertThat(indexer.getPosition(), equalTo(2));
ESTestCase.awaitBusy(() -> isFinished.get());
assertThat(indexer.getStep(), equalTo(6));
