Java 类org.elasticsearch.action.search.SearchAction 实例源码

@Override
public <Request extends ActionRequest, Response extends ActionResponse> void apply(Task task, String action,
        Request request, ActionListener<Response> listener, ActionFilterChain<Request, Response> chain) {
    if (false == action.equals(SearchAction.NAME)) {
        chain.proceed(task, action, request, listener);
        return;
    }
    if (context.getHeader(EXAMPLE_HEADER) != null) {
        throw new IllegalArgumentException("Hurray! Sent the header!");
    }
    String auth = context.getHeader(AUTHORIZATION_HEADER);
    if (auth == null) {
        ElasticsearchSecurityException e = new ElasticsearchSecurityException("Authentication required",
                RestStatus.UNAUTHORIZED);
        e.addHeader("WWW-Authenticate", "Basic realm=auth-realm");
        throw e;
    }
    if (false == REQUIRED_AUTH.equals(auth)) {
        throw new ElasticsearchSecurityException("Bad Authorization", RestStatus.FORBIDDEN);
    }
    chain.proceed(task, action, request, listener);
}

public void testCancellationDuringQueryPhase() throws Exception {

        List<ScriptedBlockPlugin> plugins = initBlockFactory();
        indexTestData();

        logger.info("Executing search");
        ListenableActionFuture<SearchResponse> searchResponse = client().prepareSearch("test").setQuery(
            scriptQuery(new Script(
                ScriptType.INLINE, "native", NativeTestScriptedBlockFactory.TEST_NATIVE_BLOCK_SCRIPT, Collections.emptyMap())))
            .execute();

        awaitForBlock(plugins);
        cancelSearch(SearchAction.NAME);
        disableBlocks(plugins);
        logger.info("Segments {}", XContentHelper.toString(client().admin().indices().prepareSegments("test").get(), FORMAT_PARAMS));
        ensureSearchWasCancelled(searchResponse);
    }

public void testCancellationDuringFetchPhase() throws Exception {

        List<ScriptedBlockPlugin> plugins = initBlockFactory();
        indexTestData();

        logger.info("Executing search");
        ListenableActionFuture<SearchResponse> searchResponse = client().prepareSearch("test")
            .addScriptField("test_field",
                new Script(ScriptType.INLINE, "native", NativeTestScriptedBlockFactory.TEST_NATIVE_BLOCK_SCRIPT, Collections.emptyMap())
            ).execute();

        awaitForBlock(plugins);
        cancelSearch(SearchAction.NAME);
        disableBlocks(plugins);
        logger.info("Segments {}", XContentHelper.toString(client().admin().indices().prepareSegments("test").get(), FORMAT_PARAMS));
        ensureSearchWasCancelled(searchResponse);
    }

public void testCancellationOfScrollSearches() throws Exception {

        List<ScriptedBlockPlugin> plugins = initBlockFactory();
        indexTestData();

        logger.info("Executing search");
        ListenableActionFuture<SearchResponse> searchResponse = client().prepareSearch("test")
            .setScroll(TimeValue.timeValueSeconds(10))
            .setSize(5)
            .setQuery(
                scriptQuery(new Script(
                    ScriptType.INLINE, "native", NativeTestScriptedBlockFactory.TEST_NATIVE_BLOCK_SCRIPT, Collections.emptyMap())))
            .execute();

        awaitForBlock(plugins);
        cancelSearch(SearchAction.NAME);
        disableBlocks(plugins);
        SearchResponse response = ensureSearchWasCancelled(searchResponse);
        if (response != null) {
            // The response might not have failed on all shards - we need to clean scroll
            logger.info("Cleaning scroll with id {}", response.getScrollId());
            client().prepareClearScroll().addScrollId(response.getScrollId()).get();
        }
    }

public void testActions() {

        // TODO this is a really shitty way to test it, we need to figure out a way to test all the client methods
        //      without specifying each one (reflection doesn't as each action needs its own special settings, without
        //      them, request validation will fail before the test is executed. (one option is to enable disabling the
        //      validation in the settings??? - ugly and conceptually wrong)

        // choosing arbitrary top level actions to test
        client.prepareGet("idx", "type", "id").execute().addListener(new AssertingActionListener<>(GetAction.NAME, client.threadPool()));
        client.prepareSearch().execute().addListener(new AssertingActionListener<>(SearchAction.NAME, client.threadPool()));
        client.prepareDelete("idx", "type", "id").execute().addListener(new AssertingActionListener<>(DeleteAction.NAME, client.threadPool()));
        client.admin().cluster().prepareDeleteStoredScript("lang", "id").execute().addListener(new AssertingActionListener<>(DeleteStoredScriptAction.NAME, client.threadPool()));
        client.prepareIndex("idx", "type", "id").setSource("source", XContentType.JSON).execute().addListener(new AssertingActionListener<>(IndexAction.NAME, client.threadPool()));

        // choosing arbitrary cluster admin actions to test
        client.admin().cluster().prepareClusterStats().execute().addListener(new AssertingActionListener<>(ClusterStatsAction.NAME, client.threadPool()));
        client.admin().cluster().prepareCreateSnapshot("repo", "bck").execute().addListener(new AssertingActionListener<>(CreateSnapshotAction.NAME, client.threadPool()));
        client.admin().cluster().prepareReroute().execute().addListener(new AssertingActionListener<>(ClusterRerouteAction.NAME, client.threadPool()));

        // choosing arbitrary indices admin actions to test
        client.admin().indices().prepareCreate("idx").execute().addListener(new AssertingActionListener<>(CreateIndexAction.NAME, client.threadPool()));
        client.admin().indices().prepareStats().execute().addListener(new AssertingActionListener<>(IndicesStatsAction.NAME, client.threadPool()));
        client.admin().indices().prepareClearCache("idx1", "idx2").execute().addListener(new AssertingActionListener<>(ClearIndicesCacheAction.NAME, client.threadPool()));
        client.admin().indices().prepareFlush().execute().addListener(new AssertingActionListener<>(FlushAction.NAME, client.threadPool()));
    }

/**
 * Perform a test search request to validate the element prior to storing it.
 *
 * @param validation validation info
 * @param element the element stored
 * @param task the parent task
 * @param listener the action listener to write to
 * @param onSuccess action ro run when the validation is successfull
 */
private void validate(FeatureValidation validation,
                      StorableElement element,
                      Task task,
                      ActionListener<FeatureStoreResponse> listener,
                      Runnable onSuccess) {
    ValidatingLtrQueryBuilder ltrBuilder = new ValidatingLtrQueryBuilder(element,
            validation, factory);
    SearchRequestBuilder builder = SearchAction.INSTANCE.newRequestBuilder(client);
    builder.setIndices(validation.getIndex());
    builder.setQuery(ltrBuilder);
    builder.setFrom(0);
    builder.setSize(20);
    // Bail out early and don't score the whole index.
    builder.setTerminateAfter(1000);
    builder.request().setParentTask(clusterService.localNode().getId(), task.getId());
    builder.execute(wrap((r) -> {
            if (r.getFailedShards() > 0) {
                ShardSearchFailure failure = r.getShardFailures()[0];
                throw new IllegalArgumentException("Validating the element caused " + r.getFailedShards() +
                        " shard failures, see root cause: " + failure.reason(), failure.getCause());
            }
            onSuccess.run();
        },
        (e) -> listener.onFailure(new IllegalArgumentException("Cannot store element, validation failed.", e))));
}

protected void executeCardinalityRequest(final NodeTaskContext context, final NodeTaskReporter reporter) {
  logger.debug("Executing async cardinality action");
  final SearchRequest cardinalityRequest = this.getCardinalityRequest(context.getNode(), context.getVisitor().getParentRequest());
  context.getClient().execute(SearchAction.INSTANCE, cardinalityRequest, new ActionListener<SearchResponse>() {

    @Override
    public void onResponse(SearchResponse searchResponse) {
      Cardinality c = searchResponse.getAggregations().get(context.getNode().getLookupPath());
      context.getNode().setCardinality(c.getValue());
      reporter.success(context);
    }

    @Override
    public void onFailure(Throwable e) {
      reporter.failure(e);
    }

  });
}

public String search(List<String> indices, List<String> types, QueryBuilder queryBuilder, Sort sort, Page page) {
    SearchRequestBuilder requestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
            .setIndices(toArray(indices))
            .setTypes(toArray(types))
            .setQuery(queryBuilder)
            .setFrom(page.offset())
            .setSize(page.size());

    sort.stream().forEach(sf -> requestBuilder.addSort(sf.field(), sf.order()));
    boolean indexExits = indices.stream().allMatch(obj -> {
        if (!indexExists(obj)) {
            return false;
        }
        return true;
    });
    SearchResponse searchResponse = null;
    if (indexExits) {
        searchResponse = client.search(requestBuilder.request())
                .actionGet();
    }
    return searchResponse != null ? searchResponse.toString() : "";

}

public HttpInvoker(Settings settings, ThreadPool threadPool, Headers headers, URL url) {
    super(settings, threadPool, headers);
    this.contexts = new HashMap<>();
    this.bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(
            Executors.newCachedThreadPool(),
            Executors.newCachedThreadPool()));
    bootstrap.setPipelineFactory(new HttpInvoker.HttpClientPipelineFactory());
    bootstrap.setOption("tcpNoDelay", true);

    registerAction(BulkAction.INSTANCE, HttpBulkAction.class);
    registerAction(CreateIndexAction.INSTANCE, HttpCreateIndexAction.class);
    registerAction(RefreshAction.INSTANCE, HttpRefreshIndexAction.class);
    registerAction(ClusterUpdateSettingsAction.INSTANCE, HttpClusterUpdateSettingsAction.class);
    registerAction(UpdateSettingsAction.INSTANCE, HttpUpdateSettingsAction.class);
    registerAction(SearchAction.INSTANCE, HttpSearchAction.class);

    this.url = url;
}

public HttpElasticsearchClient build() {
    if (url == null && host != null && port != null) {
        try {
            url = new URL("http://" + host + ":" + port);
        } catch (MalformedURLException e) {
            throw new IllegalArgumentException("malformed url: " + host + ":" + port);
        }
    }
    if (url == null) {
        throw new IllegalArgumentException("no base URL given");
    }
    ThreadPool threadpool = new ThreadPool("http_client_pool");
    client = new HttpElasticsearchClient(settings, threadpool, Headers.EMPTY, url);

    client.registerAction(BulkAction.INSTANCE, HttpBulkAction.class);
    client.registerAction(CreateIndexAction.INSTANCE, HttpCreateIndexAction.class);
    client.registerAction(RefreshAction.INSTANCE, HttpRefreshIndexAction.class);
    client.registerAction(ClusterUpdateSettingsAction.INSTANCE, HttpClusterUpdateSettingsAction.class);
    client.registerAction(UpdateSettingsAction.INSTANCE, HttpUpdateSettingsAction.class);
    client.registerAction(SearchAction.INSTANCE, HttpSearchAction.class);

    return client;
}

public Long mostRecentDocument(String index) {
    if (client() == null) {
        return null;
    }
    SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client(), SearchAction.INSTANCE);
    SortBuilder sort = SortBuilders.fieldSort("_timestamp").order(SortOrder.DESC);
    SearchResponse searchResponse = searchRequestBuilder.setIndices(index).addField("_timestamp").setSize(1).addSort(sort).execute().actionGet();
    if (searchResponse.getHits().getHits().length == 1) {
        SearchHit hit = searchResponse.getHits().getHits()[0];
        if (hit.getFields().get("_timestamp") != null) {
            return hit.getFields().get("_timestamp").getValue();
        } else {
            return 0L;
        }
    }
    return null;
}

@Override
public void execute(ActionListener<CountResponse> listener) {
    CountRequest countRequest = beforeExecute(request);
    client.execute(SearchAction.INSTANCE, countRequest.toSearchRequest(), new DelegatingActionListener<SearchResponse, CountResponse>(listener) {
        @Override
        protected CountResponse getDelegatedFromInstigator(SearchResponse response) {
            return new CountResponse(response);
        }
    });
}

public static SearchRequestBuilder constructSearchRequestBuilder(String query, Pageable p, int numberOfFragments, Client client) {

        return new SearchRequestBuilder(client, SearchAction.INSTANCE)
            .setHighlighterEncoder("html")
            .setHighlighterFragmentSize(150)
            .setHighlighterPreTags("<mark>")
            .setHighlighterPostTags("</mark>")
            .addHighlightedField("content")
            .setQuery(queryBuilder(query))
            .setFetchSource(null, "content");
    }

@Test
public void testSingleDocHttpClient() throws Exception {
    try (HttpClient client = HttpClient.builder()
            .url(new URL("http://127.0.0.1:9200"))
            .build()) {
        CreateIndexRequestBuilder createIndexRequestBuilder =
                new CreateIndexRequestBuilder(client, CreateIndexAction.INSTANCE).setIndex("test");
        createIndexRequestBuilder.execute().actionGet();
        IndexRequestBuilder indexRequestBuilder =
                new IndexRequestBuilder(client, IndexAction.INSTANCE)
                        .setIndex("test")
                        .setType("type")
                        .setId("1")
                        .setSource(jsonBuilder().startObject().field("name", "Hello World").endObject());
        indexRequestBuilder.execute().actionGet();
        RefreshRequestBuilder refreshRequestBuilder =
                new RefreshRequestBuilder(client, RefreshAction.INSTANCE)
                        .setIndices("test");
        refreshRequestBuilder.execute().actionGet();
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
        assertTrue(searchRequestBuilder.execute().actionGet().getHits().getTotalHits() > 0);
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    }
}

@Test
public void testRandomDocs() throws Exception {
    final HttpBulkClient client = HttpBulkClient.builder()
            .url(new URL("http://127.0.0.1:9200"))
            .maxActionsPerRequest(MAX_ACTIONS)
            .flushIngestInterval(TimeValue.timeValueSeconds(60))
            .build();
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", null, "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        if (client.hasException()) {
            logger.error("error", client.getException());
        }
        assertFalse(client.hasException());
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
        assertTrue(searchRequestBuilder.execute().actionGet().getHits().getTotalHits() > 0) ;
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.shutdown();
    }
}

private SearchResponse searchData(String[] urlItems, String query, int size) {

    final SearchRequestBuilder reqBuilder = new SearchRequestBuilder(
      client, SearchAction.INSTANCE);
    reqBuilder.setIndices();

    if (urlItems.length >= 1) {
      reqBuilder.setIndices(StringUtils.split(urlItems[0], ","));
    }
    if (urlItems.length > 1) {
      reqBuilder.setTypes(StringUtils.split(urlItems[1], ","));
    }

    if (!StringUtils.isEmpty(query)) {
      // The query can be either JSON-formatted, nor a Lucene query
      // So, try to parse as a JSON => if there is an error, consider the query a Lucene one
      try {
        final Map source = gson.fromJson(query, Map.class);
        reqBuilder.setExtraSource(source);
      }
      catch (JsonParseException e) {
        // This is not a JSON (or maybe not well formatted...)
        reqBuilder.setQuery(QueryBuilders.queryStringQuery(query).analyzeWildcard(true));
      }
    }

    reqBuilder.setSize(size);

    final SearchResponse response = reqBuilder.get();

    return response;
  }

@Override
public ElasticsearchResponseDto findAllBusinessObjectDefinitions(String indexName, String documentType, Set<String> facetFieldsList)
{
    LOGGER.info("Elasticsearch get all business object definition documents from index, indexName={} and documentType={}.", indexName, documentType);

    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder
        .fetchSource(new String[] {DATA_PROVIDER_NAME_SOURCE, DESCRIPTION_SOURCE, DISPLAY_NAME_SOURCE, NAME_SOURCE, NAMESPACE_CODE_SOURCE}, null);

    ElasticsearchResponseDto elasticsearchResponseDto = new ElasticsearchResponseDto();

    // Create a search request and set the scroll time and scroll size
    SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(new ElasticsearchClientImpl(), SearchAction.INSTANCE);
    searchRequestBuilder.setIndices(indexName);

    searchRequestBuilder.setTypes(documentType)
        // .setScroll(new TimeValue(ELASTIC_SEARCH_SCROLL_KEEP_ALIVE_TIME))
        //.setSize(ELASTIC_SEARCH_SCROLL_PAGE_SIZE)
        .setSource(searchSourceBuilder)

            // Set sort options.
            // First, sort on business object definition name
            // then sort on namespace code
        .addSort(SortBuilders.fieldSort(BUSINESS_OBJECT_DEFINITION_SORT_FIELD).order(SortOrder.ASC))
        .addSort(SortBuilders.fieldSort(NAMESPACE_CODE_SORT_FIELD).order(SortOrder.ASC));

    //Add aggregation builder if facet fields are present
    addFacetFieldAggregations(facetFieldsList, elasticsearchResponseDto, searchRequestBuilder, indexName);

    elasticsearchResponseDto
        .setBusinessObjectDefinitionIndexSearchResponseDtos(scrollSearchResultsIntoBusinessObjectDefinitionDto(searchRequestBuilder, indexName));

    return elasticsearchResponseDto;
}

/**
 * The ids in index function will take as arguments the index name and the document type and will return a list of all the ids in the index.
 */
@Override
public final List<String> getIdsInIndex(String indexName, String documentType)
{
    // Create an array list for storing the ids
    List<String> idList = new ArrayList<>();

    SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
    searchSourceBuilder.query(QueryBuilders.matchAllQuery());
    // Create a search request and set the scroll time and scroll size
    final SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(new ElasticsearchClientImpl(), SearchAction.INSTANCE);
    searchRequestBuilder.setIndices(indexName).setTypes(documentType).setScroll(new TimeValue(ELASTIC_SEARCH_SCROLL_KEEP_ALIVE_TIME))
        .setSize(ELASTIC_SEARCH_SCROLL_PAGE_SIZE).setSource(searchSourceBuilder);

    // Retrieve the search response
    final Search.Builder searchBuilder = new Search.Builder(searchRequestBuilder.toString()).addIndex(indexName);

    searchBuilder.setParameter(Parameters.SIZE, ELASTIC_SEARCH_SCROLL_PAGE_SIZE);
    searchBuilder.setParameter(Parameters.SCROLL, new TimeValue(ELASTIC_SEARCH_SCROLL_KEEP_ALIVE_TIME).toString());

    JestResult jestResult = jestClientHelper.searchExecute(searchBuilder.build());

    // While there are hits available, page through the results and add them to the id list
    while (jestResult.getSourceAsStringList().size() != 0)
    {
        for (String jsonString : jestResult.getSourceAsStringList())
        {
            JsonElement root = new JsonParser().parse(jsonString);
            idList.add(root.getAsJsonObject().get("id").getAsString());
        }
        String scrollId = jestResult.getJsonObject().get("_scroll_id").getAsString();
        SearchScroll scroll = new SearchScroll.Builder(scrollId, new TimeValue(ELASTIC_SEARCH_SCROLL_KEEP_ALIVE_TIME).toString()).build();
        jestResult = jestClientHelper.searchScrollExecute(scroll);

    }
    return idList;
}

@Override
public <Request extends ActionRequest, Response extends ActionResponse> void apply(final Task task, final String action, final Request request,
        final ActionListener<Response> listener, final ActionFilterChain<Request, Response> chain) {
    if (!SearchAction.INSTANCE.name().equals(action)) {
        chain.proceed(task, action, request, listener);
        return;
    }

    final SearchRequest searchRequest = (SearchRequest) request;
    final ActionListener<Response> wrappedListener = DynamicRanker.getInstance().wrapActionListener(action, searchRequest, listener);
    chain.proceed(task, action, request, wrappedListener == null ? listener : wrappedListener);
}

public SimpleRequestBuilder(ElasticsearchClient client) {
    super(client, SimpleAction.INSTANCE, new SimpleRequest());

    // here: a built-in query definition, a match all query

    SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client, SearchAction.INSTANCE);
    searchRequestBuilder
            .setQuery(QueryBuilders.matchAllQuery());

    request.setSearchRequest(searchRequestBuilder.request());
}

@Override
public void apply(final String action,
        @SuppressWarnings("rawtypes") final ActionRequest request,
        @SuppressWarnings("rawtypes") final ActionListener listener,
        final ActionFilterChain chain) {
    if (!SearchAction.INSTANCE.name().equals(action)) {
        chain.proceed(action, request, listener);
        return;
    }

    final SearchRequest searchRequest = (SearchRequest) request;
    final Boolean invoked = searchRequest.getHeader(SEARCH_REQUEST_INVOKED);
    if (invoked != null && invoked.booleanValue()) {
        if (queryResultCache.canCache(searchRequest)) {
            @SuppressWarnings({ "rawtypes", "unchecked" })
            final ActionListener cacheListener = queryResultCache
                    .execute(searchRequest, listener, chain);
            if (cacheListener != null) {
                chain.proceed(action, request, cacheListener);
            }
        } else {
            chain.proceed(action, request, listener);
        }
    } else {
        searchRequest.putHeader(SEARCH_REQUEST_INVOKED, Boolean.TRUE);
        chain.proceed(action, request, listener);
    }
}

@Test
public void testChineseLanguageCode() throws Exception {
    startCluster();
    try {
        CreateIndexRequestBuilder createIndexRequestBuilder =
                new CreateIndexRequestBuilder(client(), CreateIndexAction.INSTANCE)
                        .setIndex("test");
        createIndexRequestBuilder.addMapping("someType", jsonBuilder()
                        .startObject()
                          .startObject("properties")
                            .startObject("content")
                              .field("type", "text")
                              .startObject("fields")
                                .startObject("language")
                                   .field("type", "langdetect")
                                   .array("languages", "zh-cn")
                                .endObject()
                              .endObject()
                            .endObject()
                          .endObject()
                        .endObject());
        createIndexRequestBuilder.execute().actionGet();
        IndexRequestBuilder indexRequestBuilder = new IndexRequestBuilder(client(), IndexAction.INSTANCE)
                .setIndex("test").setType("someType").setId("1")
                .setSource("content", "位于美国首都华盛顿都会圈的希望中文学校５日晚举办活动庆祝建立２０周年。从中国大陆留学生为子女学中文而自发建立的学习班，到学生规模在全美名列前茅的中文学校，这个平台的发展也折射出美国的中文教育热度逐步提升。\n" +
                        "希望中文学校是大华盛顿地区最大中文学校，现有７个校区逾４０００名学生，规模在美国东部数一数二。不过，见证了希望中文学校２０年发展的人们起初根本无法想象这个小小的中文教育平台能发展到今日之规模。");
        indexRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).execute().actionGet();
        SearchRequestBuilder searchRequestBuilder =
                new SearchRequestBuilder(client(), SearchAction.INSTANCE)
                        .setIndices("test")
                        .setTypes("someType")
                        .setQuery(QueryBuilders.termQuery("content.language", "zh-cn"))
                        .addStoredField("content.language");
        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        assertEquals(1L, searchResponse.getHits().getTotalHits());
        assertEquals("zh-cn", searchResponse.getHits().getAt(0).getField("content.language").getValue());
    } finally {
        stopCluster();
    }
}

@Test
public void testLangDetectBinary() throws Exception {
    startCluster();
    try {
        CreateIndexRequestBuilder createIndexRequestBuilder =
                new CreateIndexRequestBuilder(client(), CreateIndexAction.INSTANCE).setIndex("test");
        createIndexRequestBuilder.addMapping("someType", jsonBuilder()
                 .startObject()
                    .startObject("properties")
                       .startObject("content")
                          .field("type", "text")
                          .startObject("fields")
                             .startObject("language")
                                .field("type", "langdetect")
                                .field("binary", true)
                             .endObject()
                          .endObject()
                       .endObject()
                    .endObject()
                .endObject());
        createIndexRequestBuilder.execute().actionGet();
        IndexRequestBuilder indexRequestBuilder =
                new IndexRequestBuilder(client(), IndexAction.INSTANCE)
                        .setIndex("test").setType("someType").setId("1")
                        //\"God Save the Queen\" (alternatively \"God Save the King\"
                        .setSource("content", "IkdvZCBTYXZlIHRoZSBRdWVlbiIgKGFsdGVybmF0aXZlbHkgIkdvZCBTYXZlIHRoZSBLaW5nIg==");
        indexRequestBuilder.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).execute().actionGet();
        SearchRequestBuilder searchRequestBuilder =
                new SearchRequestBuilder(client(), SearchAction.INSTANCE)
                        .setIndices("test")
                        .setQuery(QueryBuilders.matchAllQuery())
                        .addStoredField("content.language");
        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        assertEquals(1L, searchResponse.getHits().getTotalHits());
        assertEquals("en", searchResponse.getHits().getAt(0).getField("content.language").getValue());
    } finally {
        stopCluster();
    }
}

@Test
public void testDuplicateDocIDs() throws Exception {
    long numactions = NUM_ACTIONS;
    final BulkTransportClient client = ClientBuilder.builder()
            .put(getSettings())
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, MAX_ACTIONS)
            .setMetric(new LongAdderIngestMetric())
            .toBulkTransportClient();
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", randomString(1), "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setTypes("test")
                .setQuery(matchAllQuery());
        long hits = searchRequestBuilder.execute().actionGet().getHits().getTotalHits();
        logger.info("hits = {}", hits);
        assertTrue(hits < NUM_ACTIONS);
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.shutdown();
        assertEquals(numactions, client.getMetric().getSucceeded().getCount());
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
    }
}

@Test
public void testDuplicateDocIDs() throws Exception {
    long numactions = NUM_ACTIONS;
    final BulkNodeClient client = ClientBuilder.builder()
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, MAX_ACTIONS)
            .setMetric(new LongAdderIngestMetric())
            .toBulkNodeClient(client("1"));
    try {
        client.newIndex("test");
        for (int i = 0; i < NUM_ACTIONS; i++) {
            client.index("test", "test", randomString(1), "{ \"name\" : \"" + randomString(32) + "\"}");
        }
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setTypes("test")
                .setQuery(matchAllQuery());
        long hits = searchRequestBuilder.execute().actionGet().getHits().getTotalHits();
        logger.info("hits = {}", hits);
        assertTrue(hits < NUM_ACTIONS);
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.shutdown();
        assertEquals(numactions, client.getMetric().getSucceeded().getCount());
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
    }
}

public DeleteByQueryRequestBuilder(ElasticsearchClient client,
                                   Action<DeleteByQueryRequest, BulkByScrollResponse, DeleteByQueryRequestBuilder> action) {
    this(client, action, new SearchRequestBuilder(client, SearchAction.INSTANCE));
}

public ReindexRequestBuilder(ElasticsearchClient client,
        Action<ReindexRequest, BulkByScrollResponse, ReindexRequestBuilder> action) {
    this(client, action, new SearchRequestBuilder(client, SearchAction.INSTANCE),
            new IndexRequestBuilder(client, IndexAction.INSTANCE));
}

public UpdateByQueryRequestBuilder(ElasticsearchClient client,
        Action<UpdateByQueryRequest, BulkByScrollResponse, UpdateByQueryRequestBuilder> action) {
    this(client, action, new SearchRequestBuilder(client, SearchAction.INSTANCE));
}

@Override
public ActionFuture<SearchResponse> search(final SearchRequest request) {
    return execute(SearchAction.INSTANCE, request);
}

@Override
public void search(final SearchRequest request, final ActionListener<SearchResponse> listener) {
    execute(SearchAction.INSTANCE, request, listener);
}

@Override
public SearchRequestBuilder prepareSearch(String... indices) {
    return new SearchRequestBuilder(this, SearchAction.INSTANCE).setIndices(indices);
}

public void testSearchTaskDescriptions() {
    registerTaskManageListeners(SearchAction.NAME);  // main task
    registerTaskManageListeners(SearchAction.NAME + "[*]");  // shard task
    createIndex("test");
    ensureGreen("test"); // Make sure all shards are allocated to catch replication tasks
    client().prepareIndex("test", "doc", "test_id").setSource("{\"foo\": \"bar\"}", XContentType.JSON)
        .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();

    assertSearchResponse(client().prepareSearch("test").setTypes("doc").setQuery(QueryBuilders.matchAllQuery()).get());

    // the search operation should produce one main task
    List<TaskInfo> mainTask = findEvents(SearchAction.NAME, Tuple::v1);
    assertEquals(1, mainTask.size());
    assertThat(mainTask.get(0).getDescription(), startsWith("indices[test], types[doc], search_type["));
    assertThat(mainTask.get(0).getDescription(), containsString("\"query\":{\"match_all\""));

    // check that if we have any shard-level requests they all have non-zero length description
    List<TaskInfo> shardTasks = findEvents(SearchAction.NAME + "[*]", Tuple::v1);
    for (TaskInfo taskInfo : shardTasks) {
        assertThat(taskInfo.getParentTaskId(), notNullValue());
        assertEquals(mainTask.get(0).getTaskId(), taskInfo.getParentTaskId());
        switch (taskInfo.getAction()) {
            case SearchTransportService.QUERY_ACTION_NAME:
            case SearchTransportService.DFS_ACTION_NAME:
                assertTrue(taskInfo.getDescription(), Regex.simpleMatch("shardId[[test][*]]", taskInfo.getDescription()));
                break;
            case SearchTransportService.QUERY_ID_ACTION_NAME:
                assertTrue(taskInfo.getDescription(), Regex.simpleMatch("id[*], indices[test]", taskInfo.getDescription()));
                break;
            case SearchTransportService.FETCH_ID_ACTION_NAME:
                assertTrue(taskInfo.getDescription(), Regex.simpleMatch("id[*], size[1], lastEmittedDoc[null]",
                    taskInfo.getDescription()));
                break;
            default:
                fail("Unexpected action [" + taskInfo.getAction() + "] with description [" + taskInfo.getDescription() + "]");
        }
        // assert that all task descriptions have non-zero length
        assertThat(taskInfo.getDescription().length(), greaterThan(0));
    }

}

@Test
public void testThreadedRandomDocs() throws Exception {
    int maxthreads = Runtime.getRuntime().availableProcessors();
    int maxactions = MAX_ACTIONS;
    final long maxloop = NUM_ACTIONS;
    logger.info("HttpBulkNodeClient max={} maxactions={} maxloop={}", maxthreads, maxactions, maxloop);
    final HttpBulkClient client = HttpBulkClient.builder()
            .url(new URL("http://127.0.0.1:9200"))
            .maxActionsPerRequest(maxactions)
            .flushIngestInterval(TimeValue.timeValueSeconds(60))
            .build();
    try {
        client.newIndex("test")
                .startBulk("test", -1);
        ThreadPoolExecutor pool = EsExecutors.newFixed("http-bulk-nodeclient-test", maxthreads, 30,
                EsExecutors.daemonThreadFactory("http-bulk-nodeclient-test"));
        final CountDownLatch latch = new CountDownLatch(maxthreads);
        for (int i = 0; i < maxthreads; i++) {
            pool.execute(() -> {
                for (int j = 0; j < maxloop; j++) {
                    client.index("test", "test", null, "{ \"name\" : \"" + randomString(32) + "\"}");
                }
                latch.countDown();
            });
        }
        logger.info("waiting for max 30 seconds...");
        latch.await(30, TimeUnit.SECONDS);
        logger.info("flush...");
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        logger.info("got all responses, thread pool shutdown...");
        pool.shutdown();
        logger.info("pool is shut down");
        client.stopBulk("test", 1000);
        if (client.hasException()) {
            logger.error("error", client.getException());
        }
        assertFalse(client.hasException());
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setIndices("test")
                .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
        assertEquals(maxthreads * maxloop,
                searchRequestBuilder.execute().actionGet().getHits().getTotalHits());
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.shutdown();
    }
}

private void verifySearch(QueryBuilder query, int expectedNumOfDocs) {
    SearchRequestBuilder builder = new SearchRequestBuilder(client, SearchAction.INSTANCE);
    builder.setQuery(query);
    SearchResponse response = client.search(builder.request()).actionGet();
    assertEquals(response.getHits().getTotalHits(), expectedNumOfDocs);
}

public CoordinateSearchRequestBuilder(final ElasticsearchClient client) {
  // hack to be able to subclass SearchRequestBuilder: the action instance is only used in #execute which we overwrite
  super(client, SearchAction.INSTANCE);
}

@Override
public ActionResponse search(String[] indices, String[] types, String query, int size) {
  final SearchRequestBuilder reqBuilder = new SearchRequestBuilder(
      client, SearchAction.INSTANCE);
  reqBuilder.setIndices();

  if (indices != null) {
    reqBuilder.setIndices(indices);
  }
  if (types != null) {
    reqBuilder.setTypes(types);
  }

  if (!StringUtils.isEmpty(query)) {
    // The query can be either JSON-formatted, nor a Lucene query
    // So, try to parse as a JSON => if there is an error, consider the query a Lucene one
    try {
      @SuppressWarnings("rawtypes")
      final Map source = gson.fromJson(query, Map.class);
      reqBuilder.setExtraSource(source);
    }
    catch (final JsonSyntaxException e) {
      // This is not a JSON (or maybe not well formatted...)
      reqBuilder.setQuery(QueryBuilders.queryStringQuery(query).analyzeWildcard(true));
    }
  }

  reqBuilder.setSize(size);

  final SearchResponse searchResp = reqBuilder.get();

  final ActionResponse actionResp = new ActionResponse()
      .succeeded(true)
      .totalHits(searchResp.getHits().getTotalHits());

  if (searchResp.getAggregations() != null) {
    setAggregations(searchResp.getAggregations(), actionResp);
  }
  else {
    for (final SearchHit hit: searchResp.getHits()) {
      // Fields can be found either in _source, or in fields (it depends on the query)
      // => specific for elasticsearch's version < 5
      //
      String src = hit.getSourceAsString();
      if (src == null) {
        final Map<String, Object> hitFields = new HashMap<>();
        for (final SearchHitField hitField : hit.getFields().values()) {
          hitFields.put(hitField.getName(), hitField.getValues());
        }
        src = gson.toJson(hitFields);
      }
      actionResp.addHit(new HitWrapper(hit.getIndex(), hit.getType(), hit.getId(), src));
    }
  }

  return actionResp;
}

public SearchAction getAction() {
    return SearchAction.INSTANCE;
}

public SearchAction getAction() {
    return SearchAction.INSTANCE;
}

@Test
public void testThreadedRandomDocs() throws Exception {
    int maxthreads = Runtime.getRuntime().availableProcessors();
    long maxactions = MAX_ACTIONS;
    final long maxloop = NUM_ACTIONS;
    logger.info("HttpBulkNodeClient max={} maxactions={} maxloop={}", maxthreads, maxactions, maxloop);
    final HttpBulkNodeClient client = ClientBuilder.builder()
            .put("host", "127.0.0.1")
            .put("port", 9200)
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, maxactions)
            .put(ClientBuilder.FLUSH_INTERVAL, TimeValue.timeValueSeconds(60))
            .setMetric(new LongAdderIngestMetric())
            .toHttpBulkNodeClient();
    try {
        client.newIndex("test")
                .startBulk("test", -1, 1000);
        ThreadPoolExecutor pool = EsExecutors.newFixed("http-bulk-nodeclient-test", maxthreads, 30,
                EsExecutors.daemonThreadFactory("http-bulk-nodeclient-test"));
        final CountDownLatch latch = new CountDownLatch(maxthreads);
        for (int i = 0; i < maxthreads; i++) {
            pool.execute(new Runnable() {
                public void run() {
                    for (int i = 0; i < maxloop; i++) {
                        client.index("test", "test", null, "{ \"name\" : \"" + randomString(32) + "\"}");
                    }
                    latch.countDown();
                }
            });
        }
        logger.info("waiting for max 30 seconds...");
        latch.await(30, TimeUnit.SECONDS);
        logger.info("flush...");
        client.flushIngest();
        client.waitForResponses(TimeValue.timeValueSeconds(30));
        logger.info("got all responses, thread pool shutdown...");
        pool.shutdown();
        logger.info("pool is shut down");
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        client.stopBulk("test");
        assertEquals(maxthreads * maxloop, client.getMetric().getSucceeded().getCount());
        if (client.hasThrowable()) {
            logger.error("error", client.getThrowable());
        }
        assertFalse(client.hasThrowable());
        client.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(client.client(), SearchAction.INSTANCE)
                .setQuery(QueryBuilders.matchAllQuery()).setSize(0);
        assertEquals(maxthreads * maxloop,
                searchRequestBuilder.execute().actionGet().getHits().getTotalHits());
        client.shutdown();
    }
}

@Test
public void testThreadedRandomDocsIngestClient() throws Exception {
    int maxthreads = Runtime.getRuntime().availableProcessors();
    long maxactions = MAX_ACTIONS;
    final long maxloop = NUM_ACTIONS;
    Settings settings = Settings.settingsBuilder()
            .put("index.number_of_shards", 2)
            .put("index.number_of_replicas", 1)
            .build();
    final IngestTransportClient ingest = ClientBuilder.builder()
            .put(getSettings())
            .put(ClientBuilder.MAX_ACTIONS_PER_REQUEST, maxactions)
            .put(ClientBuilder.FLUSH_INTERVAL, TimeValue.timeValueSeconds(60))
            .setMetric(new LongAdderIngestMetric())
            .toIngestTransportClient();
    try {
        ingest.newIndex("test", settings, null)
                .startBulk("test", -1, 1000);
        ThreadPoolExecutor pool =
                EsExecutors.newFixed("ingestclient-test", maxthreads, 30, EsExecutors.daemonThreadFactory("ingestclient-test"));
        final CountDownLatch latch = new CountDownLatch(maxthreads);
        for (int i = 0; i < maxthreads; i++) {
            pool.execute(new Runnable() {
                public void run() {
                    for (int i = 0; i < maxloop; i++) {
                        ingest.index("test", "test", null, "{ \"name\" : \"" + randomString(32) + "\"}");
                    }
                    latch.countDown();
                }
            });
        }
        logger.info("waiting for max 30 seconds...");
        latch.await(30, TimeUnit.SECONDS);
        logger.info("client flush ...");
        ingest.flushIngest();
        ingest.waitForResponses(TimeValue.timeValueSeconds(30));
        logger.info("thread pool to be shut down ...");
        pool.shutdown();
        logger.info("thread pool shut down");
    } catch (NoNodeAvailableException e) {
        logger.warn("skipping, no node available");
    } finally {
        ingest.stopBulk("test");
        assertEquals(maxthreads * maxloop, ingest.getMetric().getSucceeded().getCount());
        if (ingest.hasThrowable()) {
            logger.error("error", ingest.getThrowable());
        }
        assertFalse(ingest.hasThrowable());
        ingest.refreshIndex("test");
        SearchRequestBuilder searchRequestBuilder = new SearchRequestBuilder(ingest.client(), SearchAction.INSTANCE)
                .setIndices("_all") // to avoid NPE
                .setQuery(QueryBuilders.matchAllQuery())
                .setSize(0);
        assertEquals(maxthreads * maxloop,
                searchRequestBuilder.execute().actionGet().getHits().getTotalHits());
        ingest.shutdown();
    }
}