Java 类org.elasticsearch.client.RestClient 实例源码

@Override
public void setupTest(BackendListenerContext context) throws Exception {
    try {
        this.index        = context.getParameter(ES_INDEX);
        this.bulkSize     = Integer.parseInt(context.getParameter(ES_BULK_SIZE));
        this.timeoutMs = JMeterUtils.getPropDefault(ES_TIMEOUT_MS, DEFAULT_TIMEOUT_MS);
        this.buildNumber  = (JMeterUtils.getProperty(ElasticsearchBackend.BUILD_NUMBER) != null 
                && JMeterUtils.getProperty(ElasticsearchBackend.BUILD_NUMBER).trim() != "") 
                ? Integer.parseInt(JMeterUtils.getProperty(ElasticsearchBackend.BUILD_NUMBER)) : 0;
        String host         = context.getParameter(ES_HOST);
        int port         = Integer.parseInt(context.getParameter(ES_PORT));
        this.client       = new RestHighLevelClient(
                    RestClient.builder(
                        new HttpHost(host, port, context.getParameter(ES_SCHEME, "http")))
                    .setRequestConfigCallback(requestConfigBuilder -> 
                         requestConfigBuilder
                                    .setConnectTimeout(5000)
                                    .setSocketTimeout((int)timeoutMs))
                    .setMaxRetryTimeoutMillis(60000));
        this.bulkRequest = new BulkRequest().timeout(TimeValue.timeValueMillis(timeoutMs));
        super.setupTest(context);
    } catch (Exception e) {
        throw new IllegalStateException("Unable to setup connectivity to ES", e);
    }
}

private void bulkIndexMember(List<?> memList) throws Exception {
    StringBuilder buf = new StringBuilder(1024);
    for (Object mem : memList) {
        buf.append("{\"index\": {}}");
        buf.append("\n");
        buf.append(Gson.toJson(mem));
        buf.append("\n");
    }

    long startTime = System.currentTimeMillis();
    RestClient client = Plugin.client;

    HttpEntity entity = new NStringEntity(buf.toString(), ContentType.APPLICATION_JSON);

    Response indexResponse = client.performRequest("POST", "/weike/member/_bulk",
            Collections.<String, String>emptyMap(), entity);

    if (LOG.isDebugEnabled()) {
        LOG.debug("indexMember {}ms", System.currentTimeMillis() - startTime);
        LOG.debug("indexResponse {}", indexResponse.toString());
    }
}

public void testBuildRestClient() throws Exception {
    RemoteInfo remoteInfo = new RemoteInfo("https", "localhost", 9200, new BytesArray("ignored"), null, null, emptyMap(),
            RemoteInfo.DEFAULT_SOCKET_TIMEOUT, RemoteInfo.DEFAULT_CONNECT_TIMEOUT);
    long taskId = randomLong();
    List<Thread> threads = synchronizedList(new ArrayList<>());
    RestClient client = TransportReindexAction.buildRestClient(remoteInfo, taskId, threads);
    try {
        assertBusy(() -> assertThat(threads, hasSize(2)));
        int i = 0;
        for (Thread thread : threads) {
            assertEquals("es-client-" + taskId + "-" + i, thread.getName());
            i++;
        }
    } finally {
        client.close();
    }
}

private RemoteScrollableHitSource sourceWithMockedClient(boolean mockRemoteVersion, CloseableHttpAsyncClient httpClient)
        throws Exception {
    HttpAsyncClientBuilder clientBuilder = mock(HttpAsyncClientBuilder.class);
    when(clientBuilder.build()).thenReturn(httpClient);

    RestClient restClient = RestClient.builder(new HttpHost("localhost", 9200))
            .setHttpClientConfigCallback(httpClientBuilder -> clientBuilder).build();

    TestRemoteScrollableHitSource hitSource = new TestRemoteScrollableHitSource(restClient) {
        @Override
        void lookupRemoteVersion(Consumer<Version> onVersion) {
            if (mockRemoteVersion) {
                onVersion.accept(Version.CURRENT);
            } else {
                super.lookupRemoteVersion(onVersion);
            }
        }
    };
    if (mockRemoteVersion) {
        hitSource.remoteVersion = Version.CURRENT;
    }
    return hitSource;
}

private List<Shard> buildShards(Nodes nodes, RestClient client) throws IOException {
    Response response = client.performRequest("GET", "test/_stats", singletonMap("level", "shards"));
    List<Object> shardStats = ObjectPath.createFromResponse(response).evaluate("indices.test.shards.0");
    ArrayList<Shard> shards = new ArrayList<>();
    for (Object shard : shardStats) {
        final String nodeId = ObjectPath.evaluate(shard, "routing.node");
        final Boolean primary = ObjectPath.evaluate(shard, "routing.primary");
        final Node node = nodes.getSafe(nodeId);
        final SeqNoStats seqNoStats;
        if (node.getVersion().onOrAfter(Version.V_6_0_0_alpha1_UNRELEASED)) {
            Integer maxSeqNo = ObjectPath.evaluate(shard, "seq_no.max_seq_no");
            Integer localCheckpoint = ObjectPath.evaluate(shard, "seq_no.local_checkpoint");
            Integer globalCheckpoint = ObjectPath.evaluate(shard, "seq_no.global_checkpoint");
            seqNoStats = new SeqNoStats(maxSeqNo, localCheckpoint, globalCheckpoint);
        } else {
            seqNoStats = null;
        }
        shards.add(new Shard(node, primary, seqNoStats));
    }
    return shards;
}

private static Tuple<Version, Version> readVersionsFromCatNodes(RestClient restClient) throws IOException {
    // we simply go to the _cat/nodes API and parse all versions in the cluster
    Response response = restClient.performRequest("GET", "/_cat/nodes", Collections.singletonMap("h", "version,master"));
    ClientYamlTestResponse restTestResponse = new ClientYamlTestResponse(response);
    String nodesCatResponse = restTestResponse.getBodyAsString();
    String[] split = nodesCatResponse.split("\n");
    Version version = null;
    Version masterVersion = null;
    for (String perNode : split) {
        final String[] versionAndMaster = perNode.split("\\s+");
        assert versionAndMaster.length == 2 : "invalid line: " + perNode + " length: " + versionAndMaster.length;
        final Version currentVersion = Version.fromString(versionAndMaster[0]);
        final boolean master = versionAndMaster[1].trim().equals("*");
        if (master) {
            assert masterVersion == null;
            masterVersion = currentVersion;
        }
        if (version == null) {
            version = currentVersion;
        } else if (version.onOrAfter(currentVersion)) {
            version = currentVersion;
        }
    }
    return new Tuple<>(version, masterVersion);
}

private static Version readVersionsFromInfo(RestClient restClient, int numHosts) throws IOException {
    Version version = null;
    for (int i = 0; i < numHosts; i++) {
        //we don't really use the urls here, we rely on the client doing round-robin to touch all the nodes in the cluster
        Response response = restClient.performRequest("GET", "/");
        ClientYamlTestResponse restTestResponse = new ClientYamlTestResponse(response);
        Object latestVersion = restTestResponse.evaluate("version.number");
        if (latestVersion == null) {
            throw new RuntimeException("elasticsearch version not found in the response");
        }
        final Version currentVersion = Version.fromString(latestVersion.toString());
        if (version == null) {
            version = currentVersion;
        } else if (version.onOrAfter(currentVersion)) {
            version = currentVersion;
        }
    }
    return version;
}

public ElasticBulkSender(String user, String password, HttpHost... hosts) {
    this.user = user;
    this.password = password;
    this.hosts = hosts;
    this.restClient = RestClient.builder(hosts)
            .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    if (!Strings.isBlank(user)) {
                        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                        credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(user, password));
                        return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                    } else {
                        return httpClientBuilder;
                    }
                }
            })
            .build();
}

private ReplicationResult doDeactivate(TransportContext ctx, ReplicationTransaction tx, RestClient restClient) throws ReplicationException, JSONException, IOException {
  ReplicationLog log = tx.getLog();

  ObjectMapper mapper = new ObjectMapper();
  IndexEntry content = mapper.readValue(tx.getContent().getInputStream(), IndexEntry.class);
  Response deleteResponse = restClient.performRequest(
          "DELETE",
          "/" + content.getIndex() + "/" + content.getType() + "/" + DigestUtils.md5Hex(content.getPath()),
          Collections.<String, String>emptyMap());
  LOG.debug(deleteResponse.toString());
  log.info(getClass().getSimpleName() + ": Delete Call returned " + deleteResponse.getStatusLine().getStatusCode() + ": " + deleteResponse.getStatusLine().getReasonPhrase());
  if (deleteResponse.getStatusLine().getStatusCode() == HttpStatus.SC_CREATED || deleteResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
    return ReplicationResult.OK;
  }
  LOG.error("Could not delete " + content.getType() + " at " + content.getPath());
  return new ReplicationResult(false, 0, "Replication failed");
}

/**
 * Perform the replication. All logic is covered in {@link ElasticSearchIndexContentBuilder} so we only need to transmit the JSON to ElasticSearch
 *
 * @param ctx
 * @param tx
 * @param restClient
 * @return
 * @throws ReplicationException
 */
private ReplicationResult doActivate(TransportContext ctx, ReplicationTransaction tx, RestClient restClient) throws ReplicationException, JSONException, IOException {
  ReplicationLog log = tx.getLog();
  ObjectMapper mapper = new ObjectMapper();
  IndexEntry content = mapper.readValue(tx.getContent().getInputStream(), IndexEntry.class);
  if (content != null) {
    log.info(getClass().getSimpleName() + ": Indexing " + content.getPath());
    String contentString = mapper.writeValueAsString(content.getContent());
    log.debug(getClass().getSimpleName() + ": Index-Content: " + contentString);
    LOG.debug("Index-Content: " + contentString);

    HttpEntity entity = new NStringEntity(contentString, "UTF-8");
    Response indexResponse = restClient.performRequest(
            "PUT",
            "/" + content.getIndex() + "/" + content.getType() + "/" + DigestUtils.md5Hex(content.getPath()),
            Collections.<String, String>emptyMap(),
            entity);
    LOG.debug(indexResponse.toString());
    log.info(getClass().getSimpleName() + ": " + indexResponse.getStatusLine().getStatusCode() + ": " + indexResponse.getStatusLine().getReasonPhrase());
    if (indexResponse.getStatusLine().getStatusCode() == HttpStatus.SC_CREATED || indexResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
      return ReplicationResult.OK;
    }
  }
  LOG.error("Could not replicate");
  return new ReplicationResult(false, 0, "Replication failed");
}

@Before
public void setUp() throws Exception {
    /**
     * 如果es集群安装了x-pack插件则以此种方式连接集群
     * 1. java客户端的方式是以tcp协议在9300端口上进行通信
     * 2. http客户端的方式是以http协议在9200端口上进行通信
     */
    Settings settings = Settings.builder(). put("xpack.security.user", "elastic:changeme").build();
    client = new PreBuiltXPackTransportClient(settings)
            .addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("localhost"), 9300));
    elasticsearchTemplate = new ElasticsearchTemplate(client);
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
    credentialsProvider.setCredentials(AuthScope.ANY,
            new UsernamePasswordCredentials("elastic", "changeme"));
    restClient = RestClient.builder(new HttpHost("localhost",9200,"http"))
            .setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
                @Override
                public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                    return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
            }).build();
}

@Override
    public RestClient createInstance() throws Exception {
        if (this.restClient != null) {
            return this.restClient;
        }

        HttpHost[] addresses = new HttpHost[hosts.size()];
        for (int i = 0; i < hosts.size(); i++) {
            addresses[i] = HttpHost.create(hosts.get(i));
        }

        this.restClient = RestClient
                .builder(addresses)
                .setMaxRetryTimeoutMillis(ESConstants.RESTCLIENT_TIMEOUT)
                .build();

//        this.sniffer = Sniffer.builder(this.restClient)
//                                 .setSniffIntervalMillis(60000).build();

        return this.restClient;
    }

static int getBackendVersion(ConnectionConfiguration connectionConfiguration) {
  try (RestClient restClient = connectionConfiguration.createClient()) {
    Response response = restClient.performRequest("GET", "");
    JsonNode jsonNode = parseResponse(response);
    int backendVersion = Integer
        .parseInt(jsonNode.path("version").path("number").asText().substring(0, 1));
    checkArgument((backendVersion == 2 || backendVersion == 5),
        "The Elasticsearch version to connect to is %s.x. "
        + "This version of the ElasticsearchIO is only compatible with "
        + "Elasticsearch v5.x and v2.x",
        backendVersion);
    return backendVersion;

  } catch (IOException e){
    throw (new IllegalArgumentException("Cannot get Elasticsearch version"));
  }
}

/** Inserts the given number of test documents into Elasticsearch. */
static void insertTestDocuments(ConnectionConfiguration connectionConfiguration,
    long numDocs, RestClient restClient) throws IOException {
  List<String> data =
      ElasticSearchIOTestUtils.createDocuments(
          numDocs, ElasticSearchIOTestUtils.InjectionMode.DO_NOT_INJECT_INVALID_DOCS);
  StringBuilder bulkRequest = new StringBuilder();
  int i = 0;
  for (String document : data) {
    bulkRequest.append(String.format(
        "{ \"index\" : { \"_index\" : \"%s\", \"_type\" : \"%s\", \"_id\" : \"%s\" } }%n%s%n",
        connectionConfiguration.getIndex(), connectionConfiguration.getType(), i++, document));
  }
  String endPoint = String.format("/%s/%s/_bulk", connectionConfiguration.getIndex(),
      connectionConfiguration.getType());
  HttpEntity requestBody =
      new NStringEntity(bulkRequest.toString(), ContentType.APPLICATION_JSON);
  Response response = restClient.performRequest("POST", endPoint,
      Collections.singletonMap("refresh", "true"), requestBody);
  ElasticsearchIO
      .checkForErrors(response, ElasticsearchIO.getBackendVersion(connectionConfiguration));
}

/**
 * Forces a refresh of the given index to make recently inserted documents available for search.
 *
 * @return The number of docs in the index
 */
static long refreshIndexAndGetCurrentNumDocs(
    ConnectionConfiguration connectionConfiguration, RestClient restClient) throws IOException {
  long result = 0;
  try {
    String endPoint = String.format("/%s/_refresh", connectionConfiguration.getIndex());
    restClient.performRequest("POST", endPoint);

    endPoint = String.format("/%s/%s/_search", connectionConfiguration.getIndex(),
        connectionConfiguration.getType());
    Response response = restClient.performRequest("GET", endPoint);
    JsonNode searchResult = ElasticsearchIO.parseResponse(response);
    result = searchResult.path("hits").path("total").asLong();
  } catch (IOException e) {
    // it is fine to ignore bellow exceptions because in testWriteWithBatchSize* sometimes,
    // we call upgrade before any doc have been written
    // (when there are fewer docs processed than batchSize).
    // In that cases index/type has not been created (created upon first doc insertion)
    if (!e.getMessage().contains("index_not_found_exception")){
      throw e;
    }
  }
  return result;
}

@Override
protected RestClient createInstance() throws Exception {
    HttpHost[] hosts = new HttpHost[hostnames.length];
    for (int i = 0; i < hosts.length; i++) {
        hosts[i] = HttpHost.create(hostnames[i]);
    }

    Header[] defaultHeaders = new Header[]{new BasicHeader(HEADER_CONTENT_TYPE_KEY, DEFAULT_HEADER_CONTENT_TYPE)};

    RestClient restClient = RestClient
            .builder(hosts)
            .setDefaultHeaders(defaultHeaders)
            .setFailureListener(loggingFailureListener)
            .build();

    if (enableSniffer) {
        this.sniffer = Sniffer.builder(restClient).build();
    }
    return restClient;
}

/**
 * Create an alias if needed
 * @param alias Alias name
 * @param index Index name
 * @throws Exception When alias can not be set
 */
public static void createAlias(RestClient client, String alias, String index) throws Exception {
    logger.trace("createAlias({},{})", alias, index);

    assert client != null;
    assert alias != null;
    assert index != null;

    String json = "{\"actions\":[{\"add\":{\"index\":\"" + index +"\",\"alias\":\"" + alias +"\"}}]}";

    Response response = client.performRequest("POST", "/_aliases/", Collections.<String, String>emptyMap(),
            new StringEntity(json, ContentType.APPLICATION_JSON));


    if (response.getStatusLine().getStatusCode() != 200) {
        logger.warn("Could not create alias [{}] on index [{}]", alias, index);
        throw new Exception("Could not create alias ["+alias+"] on index ["+index+"].");
    }
    logger.trace("/createAlias({},{})", alias, index);
}

/**
 * Create a new index in Elasticsearch
 * @param client Elasticsearch client
 * @param index Index name
 * @param settings Settings if any, null if no specific settings
 * @throws Exception
 */
private static void createIndexWithSettingsInElasticsearch(RestClient client, String index, String settings) throws Exception {
    logger.trace("createIndex([{}])", index);

    assert client != null;
    assert index != null;

    StringEntity entity = null;

    // If there are settings for this index, we use it. If not, using Elasticsearch defaults.
    if (settings != null) {
        logger.trace("Found settings for index [{}]: [{}]", index, settings);
        entity = new StringEntity(settings, ContentType.APPLICATION_JSON);
    }

    Response response = client.performRequest("PUT", "/" + index, Collections.<String, String>emptyMap(), entity);
    if (response.getStatusLine().getStatusCode() != 200) {
        logger.warn("Could not create index [{}]", index);
        throw new Exception("Could not create index ["+index+"].");
    }

    logger.trace("/createIndex([{}])", index);
}

/**
 * Update settings in Elasticsearch
 * @param client Elasticsearch client
 * @param index Index name
 * @param settings Settings if any, null if no update settings
 * @throws Exception
 */
private static void updateIndexWithSettingsInElasticsearch(RestClient client, String index, String settings) throws Exception {
    logger.trace("updateIndex([{}])", index);

    assert client != null;
    assert index != null;

    if (settings != null) {
        logger.trace("Found update settings for index [{}]: [{}]", index, settings);
        logger.debug("updating settings for index [{}]", index);
        StringEntity entity = new StringEntity(settings, ContentType.APPLICATION_JSON);
        client.performRequest("PUT", "/" + index + "/_settings", Collections.<String, String>emptyMap(), entity);
    }

    logger.trace("/updateIndex([{}])", index);
}

/**
 * Create a new template in Elasticsearch
 * @param client Elasticsearch client
 * @param template Template name
 * @param json JSon content for the template
 * @param force set it to true if you want to force cleaning template before adding it
 * @throws Exception
 */
public static void createTemplateWithJson(RestClient client, String template, String json, boolean force) throws Exception {
    if (isTemplateExist(client, template)) {
        if (force) {
            logger.debug("Template [{}] already exists. Force is set. Removing it.", template);
            removeTemplate(client, template);
        } else {
            logger.debug("Template [{}] already exists.", template);
        }
    }

    if (!isTemplateExist(client, template)) {
        logger.debug("Template [{}] doesn't exist. Creating it.", template);
        createTemplateWithJsonInElasticsearch(client, template, json);
    }
}

/**
 * Create a new index in Elasticsearch
 * @param client Elasticsearch client
 * @param template Template name
 * @param json JSon content for the template
 * @throws Exception
 */
private static void createTemplateWithJsonInElasticsearch(RestClient client, String template, String json) throws Exception {
    logger.trace("createTemplate([{}])", template);

    assert client != null;
    assert template != null;

    Response response = client.performRequest("PUT", "/_template/" + template, Collections.<String, String>emptyMap(),
            new StringEntity(json, ContentType.APPLICATION_JSON));


    if (response.getStatusLine().getStatusCode() != 200) {
        logger.warn("Could not create template [{}]", template);
        throw new Exception("Could not create template ["+template+"].");
    }

    logger.trace("/createTemplate([{}])", template);
}

/**
 * Create a new type in Elasticsearch
 * @param client Elasticsearch client
 * @param index Index name
 * @param type Type name
 * @param mapping Mapping if any, null if no specific mapping
 * @param merge Try to merge mapping if type already exists
 * @throws Exception
 */
public static void createMappingWithJson(RestClient client, String index, String type, String mapping, boolean merge)
        throws Exception {
    boolean mappingExist = isTypeExist(client, index, type);
    if (merge || !mappingExist) {
        if (mappingExist) {
            logger.debug("Updating type [{}]/[{}].", index, type);
        } else {
            logger.debug("Type [{}]/[{}] doesn't exist. Creating it.", index, type);
        }
        createTypeWithMappingInElasticsearch(client, index, type, mapping);
    } else {
        logger.debug("Type [{}/{}] already exists and merge is not set.", index, type);
    }

    if (mappingExist) {
        logger.debug("Type definition for [{}]/[{}] succesfully merged.", index, type);
    } else {
        logger.debug("Type definition for [{}]/[{}] succesfully created.", index, type);
    }
}

/**
 * Create a new type in Elasticsearch
 * @param client Elasticsearch client
 * @param index Index name
 * @param type Type name
 * @param mapping Mapping if any, null if no specific mapping
 * @throws Exception
 */
private static void createTypeWithMappingInElasticsearch(RestClient client, String index, String type, String mapping)
        throws Exception {
    logger.trace("createType([{}/{}])", index, type);

    assert client != null;
    assert index != null;
    assert type != null;

    if (mapping != null) {
        // Create type and mapping
        StringEntity entity = new StringEntity(mapping, ContentType.APPLICATION_JSON);
        Response response = client.performRequest("PUT", "/" + index + "/_mapping/" + type, Collections.<String, String>emptyMap(), entity);
        if (response.getStatusLine().getStatusCode() != 200) {
            logger.warn("Could not create type [{}/{}]", index, type);
            throw new Exception("Could not create type ["+index+"/"+type+"].");
        }
    } else {
        logger.trace("no content given for mapping. Ignoring type [{}/{}] creation.", index, type);
    }

    logger.trace("/createType([{}/{}])", index, type);
}

@Override
public Iterable<ValidationResult> doHealthChecks(RuntimeContainer container) {
    ValidationResult validationResult;
    try (RestClient client = ElasticsearchConnection.createClient(properties)) {
        Response response = client.performRequest("GET", "/_cluster/health", new HashMap<String, String>(),
                new BasicHeader("", ""));
        ElasticsearchResponse esResponse = new ElasticsearchResponse(response);
        if (esResponse.isOk()) {
            JsonNode entity = esResponse.getEntity();
            String status = entity.path("status").asText();
            if (status != "red") {
                validationResult = ValidationResult.OK;
            } else {
                validationResult = new ValidationResult(TalendRuntimeException.createUnexpectedException(
                        String.format("Cluster %s status is red", entity.path("cluster_name").asText())));
            }
        } else {
            validationResult = new ValidationResult(
                    TalendRuntimeException.createUnexpectedException(esResponse.getStatusLine().toString()));
        }
    } catch (IOException e) {
        validationResult = new ValidationResult(TalendRuntimeException.createUnexpectedException(e.getMessage()));
    }
    return Arrays.asList(validationResult);
}

public static RestClient createClient(ElasticsearchDatastoreProperties datastore) throws MalformedURLException {
    String urlStr = datastore.nodes.getValue();
    String[] urls = urlStr.split(",");
    HttpHost[] hosts = new HttpHost[urls.length];
    int i = 0;
    for (String address : urls) {
        URL url = new URL("http://" + address);
        hosts[i] = new HttpHost(url.getHost(), url.getPort(), url.getProtocol());
        i++;
    }
    RestClientBuilder restClientBuilder = RestClient.builder(hosts);
    if (datastore.auth.useAuth.getValue()) {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(datastore.auth.userId.getValue(), datastore.auth.password.getValue()));
        restClientBuilder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {

            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) {
                return httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
            }
        });
    }
    return restClientBuilder.build();
}

@Before
public void init() throws IOException, ExecutionException, InterruptedException {
    client = ElasticsearchTestUtils.createClient(ElasticsearchTestConstants.HOSTS.split(":")[0],
            ElasticsearchTestConstants.TRANSPORT_PORT, ElasticsearchTestConstants.CLUSTER_NAME);

    datastoreProperties = new ElasticsearchDatastoreProperties("datastore");
    datastoreProperties.init();
    datastoreProperties.nodes.setValue(ElasticsearchTestConstants.HOSTS);
    RestClient restClient = ElasticsearchConnection.createClient(datastoreProperties);

    BasicHeader emptyHeader = new BasicHeader("", "");
    Map<String, String> emptyParams = new HashMap<>();

    ElasticsearchTestUtils.deleteIndex(INDEX_NAME, client);

    Response checkExistsResponse = restClient.performRequest("HEAD", "/" + INDEX_NAME, emptyParams);
    ElasticsearchResponse checkExists = new ElasticsearchResponse(checkExistsResponse);
    if (!checkExists.isOk()) {
        // create index for test, name is 'beam'
        restClient.performRequest("PUT", "/" + INDEX_NAME, emptyHeader);
    }
}

private void indexMember(String sellerId, Object mem) throws IOException {
    if (sellerId == null)
        sellerId = "";

    long startTime = System.currentTimeMillis();

    RestClient client = Plugin.client;
    String json = Gson.toJson(mem);

    HttpEntity entity = new NStringEntity(json, ContentType.APPLICATION_JSON);
    String path = "/weike/member";
    if (!"".equals(sellerId)) {
        path += "?routing=" + sellerId;
    }
    Response indexResponse = client.performRequest("POST", path, Collections.<String, String>emptyMap(), entity);

    if (LOG.isDebugEnabled()) {
        LOG.debug("indexMember {}ms", System.currentTimeMillis() - startTime);
        LOG.debug("indexResponse {}", indexResponse.toString());
    }
}

private void startRestClient() {
    try {
        HttpHost[] hosts = new HttpHost[] {
                // new HttpHost("10.132.161.173", 30002, "http")
                new HttpHost("127.0.0.1", 9200, "http") };
        client = RestClient.builder(hosts).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
            @Override
            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                return requestConfigBuilder.setConnectTimeout(2000).setSocketTimeout(10000);
            }
        }).setMaxRetryTimeoutMillis(10000).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
            @Override
            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
                return httpClientBuilder.setMaxConnPerRoute(100).setMaxConnTotal(200);
                // return httpClientBuilder
                // .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());
            }
        }).build();
    } catch (Exception e) {
        LOG.error(e.getMessage(), e.fillInStackTrace());
    }
}

@Before
public void init() {
    HttpHost host = new HttpHost("127.0.0.1", 30002, "http");
    // HttpHost host = new HttpHost("127.0.0.1", 9002, "http");
    client = RestClient.builder(host).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
        @Override
        public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
            return requestConfigBuilder.setConnectTimeout(5000).setSocketTimeout(30000);
        }
    })
            // .setMaxRetryTimeoutMillis(30000)
            // .setHttpClientConfigCallback(new
            // RestClientBuilder.HttpClientConfigCallback() {
            // @Override
            // public HttpAsyncClientBuilder
            // customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder)
            // {
            // return httpClientBuilder
            // .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());
            // }
            // })
            .build();
}

@Before
public void init() {
    client = RestClient.builder(new HttpHost("127.0.0.1", 30002, "http"), new HttpHost("127.0.0.1", 30002, "http"))
            .setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
                @Override
                public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                    return requestConfigBuilder.setConnectTimeout(5000).setSocketTimeout(30000);
                }
            }).setMaxRetryTimeoutMillis(30000)
            // .setHttpClientConfigCallback(new
            // RestClientBuilder.HttpClientConfigCallback() {
            // @Override
            // public HttpAsyncClientBuilder
            // customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder)
            // {
            // return httpClientBuilder
            // .setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());
            // }
            // })
            .build();
}

@Before
public void setUp() throws Exception {
    final String dockerHostAddress = determineHostName();

    client = new ElasticSearchRestClient(RestClient.builder(new HttpHost(dockerHostAddress, 9200)).build()); 

    indexTweeterDocument(indexType1, 1);
    indexTweeterDocument(indexType2, 1);
    indexTweeterDocument(indexType2, 2, null);
    insertPeopleDocuments();
    indexTweeterDocument(indexType2, 1);
    indexBulkDocuments(indexName, bulkIndexType, 10);

    client.refresh(indexName);

    dataContext = new ElasticSearchRestDataContext(client, indexName);
}

@JsonCreator
public ElasticSearchFilter(@JsonProperty("hostname") String hostname, @JsonProperty("port") int port,
                           @JsonProperty("username") Optional<String> username,
                           @JsonProperty("password") Optional<String> password) {
  Header[] headers = {
    new BasicHeader(HttpHeaders.CONTENT_TYPE, "application/json"),
    new BasicHeader("Role", "Read")};
  final RestClientBuilder restClientBuilder = RestClient.builder(new HttpHost(hostname, port))
    .setDefaultHeaders(headers);
  if (username.isPresent() && !username.get().isEmpty() && password.isPresent() && !password.get().isEmpty()) {
    final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();

    credentialsProvider.setCredentials(
      AuthScope.ANY,
      new UsernamePasswordCredentials(username.get(), password.get())
    );

    restClientBuilder.setHttpClientConfigCallback(b -> b.setDefaultCredentialsProvider(credentialsProvider));
  }
  restClient = restClientBuilder.build();
  mapper = new ObjectMapper();
}

@Inject
public ESSearchService(@Named("artistDao") ArtistDao artistDao,
                       @Named("albumDao") MaprDbDao<Album> albumDao) {

    this.artistDao = artistDao;
    this.albumDao = albumDao;

    RestClient lowLevelRestClient = RestClient.builder(new HttpHost(ES_REST_HOST, ES_REST_PORT, "http")).build();
    this.client = new RestHighLevelClient(lowLevelRestClient);
}

/**
 * Build the {@link RestClient} used for reindexing from remote clusters.
 * @param remoteInfo connection information for the remote cluster
 * @param taskId the id of the current task. This is added to the thread name for easier tracking
 * @param threadCollector a list in which we collect all the threads created by the client
 */
static RestClient buildRestClient(RemoteInfo remoteInfo, long taskId, List<Thread> threadCollector) {
    Header[] clientHeaders = new Header[remoteInfo.getHeaders().size()];
    int i = 0;
    for (Map.Entry<String, String> header : remoteInfo.getHeaders().entrySet()) {
        clientHeaders[i] = new BasicHeader(header.getKey(), header.getValue());
    }
    return RestClient.builder(new HttpHost(remoteInfo.getHost(), remoteInfo.getPort(), remoteInfo.getScheme()))
            .setDefaultHeaders(clientHeaders)
            .setRequestConfigCallback(c -> {
                c.setConnectTimeout(Math.toIntExact(remoteInfo.getConnectTimeout().millis()));
                c.setSocketTimeout(Math.toIntExact(remoteInfo.getSocketTimeout().millis()));
                return c;
            })
            .setHttpClientConfigCallback(c -> {
                // Enable basic auth if it is configured
                if (remoteInfo.getUsername() != null) {
                    UsernamePasswordCredentials creds = new UsernamePasswordCredentials(remoteInfo.getUsername(),
                            remoteInfo.getPassword());
                    CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                    credentialsProvider.setCredentials(AuthScope.ANY, creds);
                    c.setDefaultCredentialsProvider(credentialsProvider);
                }
                // Stick the task id in the thread name so we can track down tasks from stack traces
                AtomicInteger threads = new AtomicInteger();
                c.setThreadFactory(r -> {
                    String name = "es-client-" + taskId + "-" + threads.getAndIncrement();
                    Thread t = new Thread(r, name);
                    threadCollector.add(t);
                    return t;
                });
                // Limit ourselves to one reactor thread because for now the search process is single threaded.
                c.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build());
                return c;
            }).build();
}

@Override
protected ScrollableHitSource buildScrollableResultSource(BackoffPolicy backoffPolicy) {
    if (mainRequest.getRemoteInfo() != null) {
        RemoteInfo remoteInfo = mainRequest.getRemoteInfo();
        createdThreads = synchronizedList(new ArrayList<>());
        RestClient restClient = buildRestClient(remoteInfo, task.getId(), createdThreads);
        return new RemoteScrollableHitSource(logger, backoffPolicy, threadPool, task::countSearchRetry, this::finishHim, restClient,
                remoteInfo.getQuery(), mainRequest.getSearchRequest());
    }
    return super.buildScrollableResultSource(backoffPolicy);
}

public RemoteScrollableHitSource(Logger logger, BackoffPolicy backoffPolicy, ThreadPool threadPool, Runnable countSearchRetry,
        Consumer<Exception> fail, RestClient client, BytesReference query, SearchRequest searchRequest) {
    super(logger, backoffPolicy, threadPool, countSearchRetry, fail);
    this.query = query;
    this.searchRequest = searchRequest;
    this.client = client;
}

@Override
protected RestClient client(String benchmarkTargetHost) {
    return RestClient
        .builder(new HttpHost(benchmarkTargetHost, 9200))
        .setHttpClientConfigCallback(b -> b.setDefaultHeaders(
            Collections.singleton(new BasicHeader(HttpHeaders.ACCEPT_ENCODING, "gzip"))))
        .setRequestConfigCallback(b -> b.setContentCompressionEnabled(true))
        .build();
}

private Task(HostsSniffer hostsSniffer, RestClient restClient, long sniffIntervalMillis, long sniffAfterFailureDelayMillis) {
    this.hostsSniffer = hostsSniffer;
    this.restClient = restClient;
    this.sniffIntervalMillis = sniffIntervalMillis;
    this.sniffAfterFailureDelayMillis = sniffAfterFailureDelayMillis;
    this.scheduledExecutorService = Executors.newScheduledThreadPool(1);
    scheduleNextRun(0);
}

public void testSniffNodes() throws IOException {
    HttpHost httpHost = new HttpHost(httpServer.getAddress().getHostString(), httpServer.getAddress().getPort());
    try (RestClient restClient = RestClient.builder(httpHost).build()) {
        ElasticsearchHostsSniffer sniffer = new ElasticsearchHostsSniffer(restClient, sniffRequestTimeout, scheme);
        try {
            List<HttpHost> sniffedHosts = sniffer.sniffHosts();
            if (sniffResponse.isFailure) {
                fail("sniffNodes should have failed");
            }
            assertThat(sniffedHosts.size(), equalTo(sniffResponse.hosts.size()));
            Iterator<HttpHost> responseHostsIterator = sniffResponse.hosts.iterator();
            for (HttpHost sniffedHost : sniffedHosts) {
                assertEquals(sniffedHost, responseHostsIterator.next());
            }
        } catch(ResponseException e) {
            Response response = e.getResponse();
            if (sniffResponse.isFailure) {
                assertThat(e.getMessage(), containsString("GET " + httpHost + "/_nodes/http?timeout=" + sniffRequestTimeout + "ms"));
                assertThat(e.getMessage(), containsString(Integer.toString(sniffResponse.nodesInfoResponseCode)));
                assertThat(response.getHost(), equalTo(httpHost));
                assertThat(response.getStatusLine().getStatusCode(), equalTo(sniffResponse.nodesInfoResponseCode));
                assertThat(response.getRequestLine().toString(),
                        equalTo("GET /_nodes/http?timeout=" + sniffRequestTimeout + "ms HTTP/1.1"));
            } else {
                fail("sniffNodes should have succeeded: " + response.getStatusLine());
            }
        }
    }
}

public void testUncompressedResponseByDefault() throws IOException {
    RestClient client = client();
    Response response = client.performRequest("GET", "/");
    assertEquals(200, response.getStatusLine().getStatusCode());
    assertNull(response.getHeader(HttpHeaders.CONTENT_ENCODING));

    response = client.performRequest("POST", "/company/employees/1", Collections.emptyMap(), SAMPLE_DOCUMENT);
    assertEquals(201, response.getStatusLine().getStatusCode());
    assertNull(response.getHeader(HttpHeaders.CONTENT_ENCODING));
}