ElasticsearchRequestSubmitter { public SearchResponse submitSearch(SearchRequest request) throws InvalidSearchException { LOG.debug("About to submit a search; request={}", ElasticsearchUtils.toJSON(request).orElse("???")); org.elasticsearch.action.search.SearchResponse esResponse; try { esResponse = client.getHighLevelClient().search(request); LOG.debug("Got Elasticsearch response with {} hit(s); response={}", esResponse.getHits().getTotalHits(), esResponse.toString()); } catch (Exception e) { String msg = String.format( "Failed to execute search; error='%s', search='%s'", ExceptionUtils.getRootCauseMessage(e), ElasticsearchUtils.toJSON(request).orElse("???")); LOG.error(msg, e); throw new InvalidSearchException(msg, e); } if(esResponse.getFailedShards() > 0) { handleShardFailures(request, esResponse); } if(RestStatus.OK == esResponse.status()) { return esResponse; } else { String msg = String.format( "Bad search response; status=%s, timeout=%s, terminatedEarly=%s", esResponse.status(), esResponse.isTimedOut(), esResponse.isTerminatedEarly()); LOG.error(msg); throw new InvalidSearchException(msg); } } ElasticsearchRequestSubmitter(ElasticsearchClient client); SearchResponse submitSearch(SearchRequest request);  }
@Test public void searchShouldSucceedWhenOK() throws InvalidSearchException, IOException { SearchResponse response = mock(SearchResponse.class); SearchRequest request = new SearchRequest(); SearchHits hits = mock(SearchHits.class); when(hits.getTotalHits()).thenReturn(1L); when(response.status()).thenReturn(RestStatus.OK); when(response.getFailedShards()).thenReturn(0); when(response.getTotalShards()).thenReturn(2); when(response.getHits()).thenReturn(hits); ElasticsearchRequestSubmitter submitter = setup(response); SearchResponse actual = submitter.submitSearch(request); assertNotNull(actual); }
@Test public void searchShouldFailWhenNotOK() throws IOException { SearchResponse response = mock(SearchResponse.class); SearchRequest request = new SearchRequest(); when(response.status()).thenReturn(RestStatus.PARTIAL_CONTENT); when(response.getFailedShards()).thenReturn(0); when(response.getTotalShards()).thenReturn(2); ElasticsearchRequestSubmitter submitter = setup(response); assertThrows(InvalidSearchException.class, () -> submitter.submitSearch(request)); }
@Test public void searchShouldHandleShardFailure() throws InvalidSearchException, IOException { SearchResponse response = mock(SearchResponse.class); SearchRequest request = new SearchRequest(); ShardSearchFailure fail = mock(ShardSearchFailure.class); SearchShardTarget target = new SearchShardTarget("node1", mock(Index.class), 1, "metron"); when(response.status()).thenReturn(RestStatus.OK); SearchHits hits = mock(SearchHits.class); when(hits.getTotalHits()).thenReturn(1L); when(response.getFailedShards()).thenReturn(1); when(response.getTotalShards()).thenReturn(2); when(response.getHits()).thenReturn(hits); ShardSearchFailure[] failures = { fail }; when(response.getShardFailures()).thenReturn(failures); when(fail.shard()).thenReturn(target); when(fail.index()).thenReturn("bro_index_2017-10-11"); when(fail.shardId()).thenReturn(1); ElasticsearchRequestSubmitter submitter = setup(response); SearchResponse actual = submitter.submitSearch(request); assertNotNull(actual); }