honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-02-17 18:35:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

[TEST] Make BulkTests#testBulkProcessorFlush more robust

Browse Source
This commit is contained in:
Simon Willnauer 2014-04-03 13:32:41 +02:00
parent 6f5b7fa086
commit 697432390d
1 changed files with 17 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
src/test/java/org/elasticsearch/document/BulkTests.java
View File

@ -40,12 +40,9 @@ import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.test.ElasticsearchIntegrationTest; import org.elasticsearch.test.ElasticsearchIntegrationTest;
import org.junit.Test; import org.junit.Test;
import java.util.Arrays;
import java.util.Map; import java.util.Map;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.*;
import java.util.concurrent.CyclicBarrier; import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder; import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*; import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.*;
@ -624,19 +621,24 @@ public class BulkTests extends ElasticsearchIntegrationTest {
@Test @Test
public void testBulkProcessorFlush() throws InterruptedException { public void testBulkProcessorFlush() throws InterruptedException {
final BlockingQueue<BulkResponse> responseQueue = new SynchronousQueue(); final AtomicReference<BulkResponse> responseRef = new AtomicReference<>();
final AtomicReference<Throwable> failureRef = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
BulkProcessor.Listener listener = new BulkProcessor.Listener() { BulkProcessor.Listener listener = new BulkProcessor.Listener() {
@Override @Override
public void beforeBulk(long executionId, BulkRequest request) {} public void beforeBulk(long executionId, BulkRequest request) {}
@Override @Override
public void afterBulk(long executionId, BulkRequest request, BulkResponse response) { public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
responseQueue.add(response); responseRef.set(response);
latch.countDown();
} }
@Override @Override
public void afterBulk(long executionId, BulkRequest request, Throwable failure) {} public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
failureRef.set(failure);
latch.countDown();
}
}; };
try (BulkProcessor processor = BulkProcessor.builder(client(), listener).setBulkActions(6) try (BulkProcessor processor = BulkProcessor.builder(client(), listener).setBulkActions(6)
@ -651,10 +653,13 @@ public class BulkTests extends ElasticsearchIntegrationTest {
processor.add(new IndexRequest("test", "test", "5").source(data)); processor.add(new IndexRequest("test", "test", "5").source(data));
processor.flush(); processor.flush();
latch.await();
BulkResponse response = responseQueue.poll(1, TimeUnit.SECONDS); BulkResponse response = responseRef.get();
assertThat("Could not get a bulk response even after an explicit flush.", response, is(notNullValue())); Throwable error = failureRef.get();
assertThat(error, nullValue());
assertThat("Could not get a bulk response even after an explicit flush.", response, notNullValue());
assertThat(response.getItems().length, is(5)); assertThat(response.getItems().length, is(5));
} }
} }
} }