BatchSearchRunner implements Callable<Integer>, Monitorable, UserTask { @Override public Integer call() { List<BatchSearch> batchSearches = repository.getQueued(); totalNbBatches = batchSearches.size(); logger.info("found {} queued batch searches", batchSearches.size()); int totalResults = 0; for (BatchSearch batchSearch : batchSearches) { totalResults += run(batchSearch); totalProcessed += 1; } logger.info("done {} batch searches", batchSearches.size()); return totalResults; } @Inject  BatchSearchRunner(Indexer indexer, BatchSearchRepository repository, PropertiesProvider propertiesProvider, @Assisted User user); @Override Integer call(); int run(String batchSearchId); @Override double getProgressRate(); @Override User getUser(); void close();  }
@Test public void test_run_batch_searches() throws Exception { Document[] documents = {createDoc("doc1").build(), createDoc("doc2").build()}; firstSearchWillReturn(1, documents); when(repository.getQueued()).thenReturn(asList( new BatchSearch("uuid1", project("test-datashare"), "name1", "desc1", asSet("query1", "query2"), new Date(), BatchSearch.State.RUNNING, User.local()), new BatchSearch("uuid2", project("test-datashare"), "name2", "desc1", asSet("query3", "query4"), new Date(), BatchSearch.State.RUNNING, User.local()) )); assertThat(new BatchSearchRunner(indexer, repository, new PropertiesProvider(), local()).call()).isEqualTo(2); verify(repository).saveResults("uuid1", "query1", asList(documents)); verify(repository).setState("uuid1", BatchSearch.State.RUNNING); verify(repository).setState("uuid1", BatchSearch.State.SUCCESS); verify(repository, never()).saveResults(eq("uuid2"), anyString(), anyList()); }
@Test public void test_run_batch_search_failure() throws Exception { Document[] documents = {createDoc("doc").build()}; firstSearchWillReturn(1, documents); when(repository.getQueued()).thenReturn(singletonList( new BatchSearch("uuid1", project("test-datashare"), "name1", "desc1", asSet("query1", "query2"), new Date(), BatchSearch.State.RUNNING, User.local()) )); when(repository.saveResults(anyString(), any(), anyList())).thenThrow(new RuntimeException()); assertThat(new BatchSearchRunner(indexer, repository, new PropertiesProvider(), local()).call()).isEqualTo(0); verify(repository).setState("uuid1", BatchSearch.State.RUNNING); verify(repository).setState(eq("uuid1"), any(SearchException.class)); }
@Test public void test_run_batch_search_truncate_to_60k_max_results() throws Exception { Document[] documents = IntStream.range(0, MAX_SCROLL_SIZE).mapToObj(i -> createDoc("doc" + i).build()).toArray(Document[]::new); firstSearchWillReturn(MAX_BATCH_RESULT_SIZE/MAX_SCROLL_SIZE + 1, documents); when(repository.getQueued()).thenReturn(singletonList( new BatchSearch("uuid1", project("test-datashare"), "name", "desc", asSet("query"), new Date(), BatchSearch.State.RUNNING, User.local()) )); assertThat(new BatchSearchRunner(indexer, repository, new PropertiesProvider(), local()).call()).isLessThan(60000); }
@Test public void test_run_batch_search_with_throttle() throws Exception { firstSearchWillReturn(1, createDoc("doc").build()); when(repository.getQueued()).thenReturn(singletonList( new BatchSearch("uuid1", project("test-datashare"), "name1", "desc1", asSet("query1", "query2"), new Date(), BatchSearch.State.RUNNING, User.local()) )); Date beforeBatch = timeRule.now; new BatchSearchRunner(indexer, repository, new PropertiesProvider(new HashMap<String, String>() {{ put(BATCH_SEARCH_THROTTLE, "1000"); }}), local()).call(); assertThat(timeRule.now().getTime() - beforeBatch.getTime()).isEqualTo(1000); }
@Test public void test_run_batch_search_with_throttle_should_not_last_more_than_max_time() throws Exception { firstSearchWillReturn(5, createDoc("doc").build()); when(repository.getQueued()).thenReturn(singletonList( new BatchSearch("uuid1", project("test-datashare"), "name1", "desc1", asSet("query1", "query2"), new Date(), BatchSearch.State.RUNNING, User.local()) )); Date beforeBatch = timeRule.now; new BatchSearchRunner(indexer, repository, new PropertiesProvider(new HashMap<String, String>() {{ put(BATCH_SEARCH_THROTTLE, "1000"); put(BATCH_SEARCH_MAX_TIME, "1"); }}), local()).call(); assertThat(timeRule.now().getTime() - beforeBatch.getTime()).isEqualTo(1000); }