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

ClientYamlTestResponse(Response response) throws IOException {
    this.response = response;
    if (response.getEntity() != null) {
        String contentType = response.getHeader("Content-Type");
        this.bodyContentType = XContentType.fromMediaTypeOrFormat(contentType);
        try {
            byte[] bytes = EntityUtils.toByteArray(response.getEntity());
            //skip parsing if we got text back (e.g. if we called _cat apis)
            if (bodyContentType != null) {
                this.parsedResponse = ObjectPath.createFromXContent(bodyContentType.xContent(), new BytesArray(bytes));
            }
            this.body = bytes;
        } catch (IOException e) {
            EntityUtils.consumeQuietly(response.getEntity());
            throw e;
        }
    } else {
        this.body = null;
        this.bodyContentType = null;
    }
}

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());
    }
}

private ActionFuture<String> start(String method, String path, HttpEntity body) {
    PlainActionFuture<String> future = new PlainActionFuture<>();
    Map<String, String> params = new HashMap<>();
    params.put("refresh", "wait_for");
    params.put("error_trace", "");
    client().performRequestAsync(method, docPath() + path, params, body, new ResponseListener() {
        @Override
        public void onSuccess(Response response) {
            try {
                future.onResponse(EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8));
            } catch (IOException e) {
                future.onFailure(e);
            }
        }

        @Override
        public void onFailure(Exception exception) {
            future.onFailure(exception);
        }
    });
    return future;
}

@Override
public boolean bulkIndex(List<String> bulkData) {
    StringBuilder bulkRequestBody = new StringBuilder();
    for (String bulkItem : bulkData) {
        bulkRequestBody.append(actionMetaData);
        bulkRequestBody.append(bulkItem);
        bulkRequestBody.append("\n");
    }
    HttpEntity entity = new NStringEntity(bulkRequestBody.toString(), ContentType.APPLICATION_JSON);
    try {
        Response response = client.performRequest("POST", "/geonames/type/_noop_bulk", Collections.emptyMap(), entity);
        return response.getStatusLine().getStatusCode() == HttpStatus.SC_OK;
    } catch (Exception e) {
        throw new ElasticsearchException(e);
    }
}

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 Nodes buildNodeAndVersions() throws IOException {
    Response response = client().performRequest("GET", "_nodes");
    ObjectPath objectPath = ObjectPath.createFromResponse(response);
    Map<String, Object> nodesAsMap = objectPath.evaluate("nodes");
    Nodes nodes = new Nodes();
    for (String id : nodesAsMap.keySet()) {
        nodes.add(new Node(
            id,
            objectPath.evaluate("nodes." + id + ".name"),
            Version.fromString(objectPath.evaluate("nodes." + id + ".version")),
            HttpHost.create(objectPath.evaluate("nodes." + id + ".http.publish_address"))));
    }
    response = client().performRequest("GET", "_cluster/state");
    nodes.setMasterNodeId(ObjectPath.createFromResponse(response).evaluate("master_node"));
    return nodes;
}

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;
}

/**
     * Probably there is a better way to get the index name with the alias.
     * If there is no such index, than null is returned.
     * @param aliasName
     * @return
     */
    public String findIndex(String aliasName) {
        String indexName = null;
        try {
            Response response = this.restClient.performRequest("HEAD", aliasName);
            if (response.getStatusLine().getStatusCode() != 404) {
                // Index with aliasName exists, so we need to get real index name from settings
                response = this.restClient.performRequest("GET", aliasName.concat("/_settings"));
//                JSONObject json = new JSONObject(EntityUtils.toString(response.getEntity()));
//                indexName = json.keySet().stream().findFirst().orElse("");
            }
        } catch (IOException e) {
            LOG.debug("OK. No index with a given alias exists.");
        }
        return indexName;
    }

private void createIndex() {
  Response response;

  try (InputStream payload = FactSearchManager.class.getClassLoader().getResourceAsStream(MAPPINGS_JSON)) {
    // Need to use low-level client here because the Index API is not yet supported by the high-level client.
    HttpEntity body = new InputStreamEntity(payload, ContentType.APPLICATION_JSON);
    response = clientFactory.getLowLevelClient().performRequest("PUT", INDEX_NAME, Collections.emptyMap(), body);
  } catch (IOException ex) {
    throw logAndExit(ex, "Could not perform request to create index.");
  }

  if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
    String msg = String.format("Could not create index '%s'.", INDEX_NAME);
    LOGGER.error(msg);
    throw new IllegalStateException(msg);
  }

  LOGGER.info("Successfully created index '%s'.", INDEX_NAME);
}

/**
 * Creates the specified index in ElasticSearch
 * 
 * @param indexName
 *            the index name to augment
 * @param typeName
 *            the type name to augment
 * @param id
 *            the id of the document to add
 * @param jsonDocument
 *            the String JSON document to add
 */
protected static void createIndex(String indexName) {
    try {
        // Create our expand / search indices
        String endpoint = String.format("/%s", indexName);
        Map<String, String> params = new HashMap<String, String>();
        StringEntity requestBody = new StringEntity(INDEX_JSON);

        Response resp = client.performRequest("PUT", endpoint, params, requestBody, contentTypeHeader);
        staticLogger.debug("Response: " + resp.getStatusLine());

    } catch (IOException e) {
        // Ignore this...? probably already exists
        staticLogger.error(e.getMessage(), e);

        if (e instanceof UnsupportedEncodingException) {
            staticLogger.error("Error encoding JSON: " + e.getMessage(), e);
            return;
        }
    }
}

/**
 * Adds a document to the specified ElasticSearch index / type
 * 
 * @param indexName
 *            the index name to augment
 * @param typeName
 *            the type name to augment
 * @param id
 *            the id of the document to add
 * @param jsonDocument
 *            the String JSON document to add
 */
protected static void addDocument(String indexName, String typeName, Integer id, String jsonDocument) {
    try {
        String documentEndpoint = String.format("/%s/%s/%d", indexName, typeName, id);
        StringEntity requestBody = new StringEntity(jsonDocument);
        Map<String, String> params = new HashMap<String, String>();

        Response resp = client.performRequest("PUT", documentEndpoint, params, requestBody, contentTypeHeader);
        staticLogger.debug("Response: " + resp.getStatusLine());

    } catch (IOException e) {
        // Ignore this...? probably already exists
        staticLogger.error(e.getMessage(), e);

        if (e instanceof UnsupportedEncodingException) {
            staticLogger.error("Error encoding JSON: " + e.getMessage(), e);
            return;
        }
    }
}

@Test
@SuppressWarnings("unchecked")
public void testPluginIsLoaded() throws Exception {

    Response response = client.performRequest("GET", "/_nodes/plugins");

    Map<String, Object> nodes = (Map<String, Object>) entityAsMap(response).get("nodes");
    for (String nodeName : nodes.keySet()) {
        boolean pluginFound = false;
        Map<String, Object> node = (Map<String, Object>) nodes.get(nodeName);
        List<Map<String, Object>> plugins = (List<Map<String, Object>>) node.get("plugins");
        for (Map<String, Object> plugin : plugins) {
            String pluginName = (String) plugin.get("name");
            if (pluginName.equals("rocchio")) {
                pluginFound = true;
                break;
            }
        }
        assertThat(pluginFound, is(true));
    }
}

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");
}

public SearchResult perform() {
  try {
    ObjectMapper mapper = new ObjectMapper();
    String queryString = serialize();
    LOG.info("Query: " + queryString);
    Response response = elasticSearchService.getRestClient().performRequest(
            "GET",
            "/" + index + "/_search",
            Collections.<String, String>emptyMap(),
            new StringEntity(queryString, "UTF-8"));

    HttpEntity entity = response.getEntity();
    return mapper.readValue(entity.getContent(), SearchResult.class);
  }
  catch (IOException ex) {
    LOG.error("Could not perform query", ex);
  }
  return null;
}

/**
 * Retrieve the informatoin of a realm
 * @param bot The bot instance.
 * @param region The region the realm is in
 * @param realm The realm name
 * @return A Json string containing information about the realm. Returns null if no realm is found.
 */
public static String getRealmInformation(LegendaryBot bot, String region, String realm) {
    HttpEntity entity = new NStringEntity("{ \"query\": { \"match\" : { \"name\" : \""+realm+"\" } } }", ContentType.APPLICATION_JSON);
    try {
        Response response = bot.getElasticSearch().performRequest("POST", "/wow/realm_"+region.toLowerCase()+"/_search", Collections.emptyMap(), entity);
        String jsonResponse = EntityUtils.toString(response.getEntity());
        JSONParser jsonParser = new JSONParser();
        JSONObject obj = (JSONObject) jsonParser.parse(jsonResponse);
        JSONArray hit = (JSONArray) ((JSONObject)obj.get("hits")).get("hits");
        if (hit.size() == 0) {
            return null;
        }
        JSONObject firstItem = (JSONObject) hit.get(0);
        JSONObject source = (JSONObject)  firstItem.get("_source");
        return source.toJSONString();
    } catch (IOException | ParseException e) {
        e.printStackTrace();
    }
    return null;
}

public void testIndexTweet() throws IOException {
    MarcElasticsearchClient client = new MarcElasticsearchClient();
    Response response = client.indexTweet(2, "kimchy", "trying out Elasticsearch");
    assertEquals("HTTP/1.1 201 Created", response.getStatusLine().toString());
    assertEquals(3, response.getHeaders().length);

    assertEquals("Location", response.getHeaders()[0].getName());
    assertEquals("/twitter/tweet/2", response.getHeaders()[0].getValue());
    assertEquals("content-type", response.getHeaders()[1].getName());
    assertEquals("application/json; charset=UTF-8", response.getHeaders()[1].getValue());
    assertEquals("content-length", response.getHeaders()[2].getName());
    assertTrue(140 < Integer.parseInt(response.getHeaders()[2].getValue()));

    String json = EntityUtils.toString(response.getEntity());
    // {"_index":"twitter","_type":"tweet","_id":"2","_version":161,"result":"created","_shards":{"total":2,"successful":1,"failed":0},"created":true}
    assertTrue(json.startsWith("{\"_index\":\"twitter\",\"_type\":\"tweet\",\"_id\":\"2\",\"_version\":"));
    assertTrue(json.endsWith(",\"result\":\"created\",\"_shards\":{\"total\":2,\"successful\":1,\"failed\":0},\"created\":true}"));

    // assertEquals(2, client.getNumberOfTweets());
}

public void testDeleteTweet() throws IOException {
    MarcElasticsearchClient client = new MarcElasticsearchClient();
    Response response = client.deleteTweet(2);
    assertEquals(org.apache.http.message.BasicStatusLine.class, response.getStatusLine().getClass());
    assertEquals("HTTP/1.1 200 OK", response.getStatusLine().toString());
    assertEquals(2, response.getHeaders().length);
    assertEquals("content-type", response.getHeaders()[0].getName());
    assertEquals("application/json; charset=UTF-8", response.getHeaders()[0].getValue());
    assertEquals("content-length", response.getHeaders()[1].getName());
    assertTrue(130 < Integer.parseInt(response.getHeaders()[1].getValue()));
    String json = EntityUtils.toString(response.getEntity());
    // '{"found":true,"_index":"twitter","_type":"tweet","_id":"2","_version":156,"result":"deleted","_shards":{"total":2,"successful":1,"failed":0}}'
    assertTrue(json.startsWith("{\"found\":true,\"_index\":\"twitter\",\"_type\":\"tweet\",\"_id\":\"2\",\"_version\":"));
    assertTrue(json.endsWith(",\"result\":\"deleted\",\"_shards\":{\"total\":2,\"successful\":1,\"failed\":0}}"));
    assertEquals(2, client.getNumberOfTweets());
}

public GetResponseProjection get(String index, String type, String id) {
    StringBuilder sb = new StringBuilder();
    sb
            .append(index)
            .append("/").append(type)
            .append("/").append(id);
    Response response = esRestClient.performRequest("GET", sb.toString());

    try {
        String s = EntityUtils.toString(response.getEntity());
        GetResponseProjection responseProjection = JsonUtils.parseJsonToObject(s, GetResponseProjection.class);

        return responseProjection;
    } catch (IOException e) {
        throw new SearchException(e);
    }
}

public ESBulkResponse doBulkRequest(String body) {
    try {
        HttpEntity requestBody = new StringEntity(body);
        Response response = client.performRequest(
                "POST",
                ESConstants.STORE_INDEX + "/" + ESConstants.PRODUCT_TYPE + "/_bulk",
                new HashMap<String, String>(),
                requestBody);

        ESBulkResponse esResponse = gson.fromJson(IOUtils.toString(response.getEntity().getContent()),
                ESBulkResponse.class);
        return esResponse;
    } catch (IOException e) {
        logger.error("Error bulk request " + e);
    }

    return null;
}

public ESMultiGetResponse multiGet(HashMap<String, String> params, String requestBody) {
    try {
        HttpEntity requestEntity = new StringEntity(requestBody);
        Response response = client.performRequest(
                "POST",
                ESConstants.STORE_INDEX + "/" + ESConstants.SKU_TYPE + "/_mget",
                params,
                requestEntity);
        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode != 200) {
            logger.warn("Not found" + response.toString());
            return null;
        }
        String resStr = IOUtils.toString(response.getEntity().getContent());
        return gson.fromJson(resStr, ESMultiGetResponse.class);
    } catch (IOException e) {
        logger.error("Failed to get document with type  " + e);
    }

    return null;
}

public ESQueryResponse searchAll(Map<String, String> params, String body) {
    try {
        HttpEntity entity = new StringEntity(body);
        Response response = client.performRequest(
                "GET",
                ESConstants.STORE_INDEX  + "/_search",
                params,
                entity);

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode > 299) {
            logger.warn("Problem while search a document: {}", response.getStatusLine().getReasonPhrase());
            return null;
        }

        ESQueryResponse esQueryResponse = gson.fromJson(IOUtils.toString(response.getEntity().getContent()),
                ESQueryResponse.class);
        return esQueryResponse;
    } catch (IOException e) {
        logger.error("update failed " + e);
    }

    return null;
}

public ESQueryResponse searchScroll(Map<String, String> params, String body) {
    try {
        HttpEntity entity = new StringEntity(body);
        Response response = client.performRequest(
                "POST",
                "/_search/scroll",
                params,
                entity);

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode > 299) {
            logger.warn("Problem while indexing a document: {}", response.getStatusLine().getReasonPhrase());
            return null;
        }

        ESQueryResponse esQueryResponse = gson.fromJson(IOUtils.toString(response.getEntity().getContent()),
                ESQueryResponse.class);
        return esQueryResponse;
    } catch (IOException e) {
        logger.error("update failed " + e);
    }

    return null;
}

public boolean isDocExist(String index, String type, Long sourceId, HashMap param) {
    Map<String, String> params = addTenantId2Param(param);
    try {
        Response response = client.performRequest("HEAD", index + "/" + type + "/" + sourceId, params);
        int statusCode = response.getStatusLine().getStatusCode();

        if (statusCode == 200) {
            return true;
        } else if (statusCode == 404) {
            return false;
        }

        throw new ElasticQueryException("Exception for Doc Exist query");
    } catch (IOException e) {
        logger.warn("Failed to verify the index existence ", e);
        throw new ElasticAPIException("Call indexExist HEAD exception:"+e.getMessage());
    }
}

public ESBulkResponse doBulkRequest(String index, String type, Map<String, String> params, String body) {
    params = addTenantId2Param(params);
    try {
        HttpEntity requestBody = new StringEntity(body);
        Response response = client.performRequest(
                "POST",
                index + "/" + type + "/_bulk",
                params,
                requestBody);

        ESBulkResponse esResponse = gson.fromJson(IOUtils.toString(response.getEntity().getContent()),
                ESBulkResponse.class);
        return esResponse;
    } catch (IOException e) {
        logger.error("Failed to delete document with type [" + type +  "]" + e);
    }

    return null;
}

public ESDeleteResponse Delete(String index, String type, Long sourceId, Map<String, String> params) {
    params = addTenantId2Param(params);
    try {
        Response response = client.performRequest(
                "DELETE",
                index + "/" + type + "/" + sourceId,
                params);

        int statusCode = response.getStatusLine().getStatusCode();

        if (statusCode > 299) {
            logger.warn("Problem while indexing a document: {}" + response.getStatusLine().getReasonPhrase());
            throw new ElasticAPIException("Could not index a document, status code is " + statusCode);
        }

        ESDeleteResponse deleteResponse = gson.fromJson(IOUtils.toString(response.getEntity().getContent()),
                ESDeleteResponse.class);
        return deleteResponse;
    } catch (IOException e) {
        logger.error("Failed to delete document with type [" + type + "] id ["
                + sourceId + "]: ",e);
    }
    return null;
}

public ESMultiGetResponse multiGet(String index, String type, HashMap<String, String> params, String requestBody) {
    Map<String, String> param = addTenantId2Param(params);

    try {
        HttpEntity requestEntity = new StringEntity(requestBody);
        Response response = client.performRequest(
                "POST",
                index + "/" + type + "/_mget",
                param,
                requestEntity);
        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode != 200) {
            logger.warn("Not found" + response.toString());
            return null;
        }

        logger.info("Got response :{}" + response.getEntity().toString());
        String resStr = IOUtils.toString(response.getEntity().getContent());
        return gson.fromJson(resStr, ESMultiGetResponse.class);
    } catch (IOException e) {
        logger.error("Failed to get document with type [" + type + "] ",e);
    }

    return null;
}

public Long count(String index, String type, String query) {
    Map<String, String> params = addTenantId2Param(null);
    try {
        Response response = client.performRequest(
                "GET",
                index + "/" + type + "/_count",
                params,
                new StringEntity(query));

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode !=200 ) {
            logger.warn("Problem while indexing a document: {}" + response.getStatusLine().getReasonPhrase());
            throw new ElasticAPIException("Could not index a document, status code is " + statusCode);
        }

        ESCountResponse countResponse = gson.fromJson(IOUtils.toString(response.getEntity().getContent()),
                ESCountResponse.class);
        if (countResponse != null) {
           return countResponse.getCount();
        }
    } catch (IOException e) {
        logger.error("Failed to count document with type ["+ type + "]: ",e);
    }
    return null;
}

public JsonObject loadSourceObjectById(String index, String type, Long sourceId, HashMap param) {
    Map<String, String> params = addTenantId2Param(param);
    try {
        Response response = client.performRequest(
                "GET",
                index + "/" + type + "/" + sourceId + "/_source",
                params);

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode != 200) {
            logger.warn("Problem while indexing a document: {}" + response.getStatusLine().getReasonPhrase());
            return null;
        }
        return gson.fromJson(IOUtils.toString(response.getEntity().getContent()), JsonObject.class);
    } catch (IOException e) {
        logger.error("Failed to get document with type [" + type + "] id [" + sourceId + "]: ",e);
    }
    return null;
}

public ESQueryResponse query(String index, String type, Map<String, String> params, String body) {
    params = addTenantId2Param(params);

    HttpEntity requestBody = new StringEntity(body, Charsets.UTF_8);
    try {
        Response response = client.performRequest(
                "GET",
                index + "/" + type + "/_search",
                params,
                requestBody);

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode > 299) {
            logger.warn("Problem while indexing a document: {}" + response.getStatusLine().getReasonPhrase());
            throw new ElasticAPIException("Could not index a document, status code is " + statusCode);
        }

        ESQueryResponse esQueryResponse = gson.fromJson(IOUtils.toString(response.getEntity().getContent()),
                ESQueryResponse.class);

        return esQueryResponse;
    } catch (IOException e) {
        logger.error("Failed to update document with type [" + type + "] body [" + body + "]: ",e);
    }
    return null;
}

public Boolean indexExist(String indexName) {
    try {
        Response response = client.performRequest("HEAD", indexName);

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode == 200) {
            return true;
        } else if (statusCode == 404) {
            return false;
        }
        logger.error("Error checking index existence: {}" + response.getStatusLine().getReasonPhrase());

    } catch (IOException e) {
        logger.error("Failed to verify the index existence ", e);
    }

    return true;
}

public String search(String q) throws IOException {
    HttpEntity entity1 = new NStringEntity(
            "{\n" +
                    "    \"query\" : {\n" +
                    "    \"match\": { \"dc\":\""+q+"\"} \n" +
                    "}, \n"+
                    "    \"sort\" : [\n" +
                    "    {\"title.keyword\": { \"order\":\"desc\"}} \n" +
                    "], \n"+
                    "\"_source\":\"title\""+
                    "}"
            , ContentType.APPLICATION_JSON);

    Response response = restClient.performRequest("GET", "/harvester/_search", Collections.singletonMap("pretty", "true"),
            entity1);
    String result =  ( EntityUtils.toString(response.getEntity()));
    System.out.println(result);

    return "<pre>"+result+"</pre>"; //pre tag for json, otherwise it didnt show pretty in browser

}

@Override
public boolean advance() throws IOException {
  if (batchIterator.hasNext()) {
    current = batchIterator.next();
    return true;
  } else {
    String requestBody =
        String.format(
            "{\"scroll\" : \"%s\",\"scroll_id\" : \"%s\"}",
            source.spec.getScrollKeepalive(), scrollId);
    HttpEntity scrollEntity = new NStringEntity(requestBody, ContentType.APPLICATION_JSON);
    Response response =
        restClient.performRequest(
            "GET",
            "/_search/scroll",
            Collections.<String, String>emptyMap(),
            scrollEntity);
    JsonNode searchResult = parseResponse(response);
    updateScrollId(searchResult);
    return readNextBatchAndReturnFirstDocument(searchResult);
  }
}

private void flushBatch() throws IOException {
  if (batch.isEmpty()) {
    return;
  }
  StringBuilder bulkRequest = new StringBuilder();
  for (String json : batch) {
    bulkRequest.append(json);
  }
  batch.clear();
  currentBatchSizeBytes = 0;
  Response response;
  String endPoint =
      String.format(
          "/%s/%s/_bulk",
          spec.getConnectionConfiguration().getIndex(),
          spec.getConnectionConfiguration().getType());
  HttpEntity requestBody =
      new NStringEntity(bulkRequest.toString(), ContentType.APPLICATION_JSON);
  response =
      restClient.performRequest(
          "POST",
          endPoint,
          Collections.<String, String>emptyMap(),
          requestBody);
  checkForErrors(response, backendVersion);
}

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;
}

public Boolean indexExist(String indexName) {
    try {
        Response response = client.performRequest(
                "HEAD",
                indexName
        );

        int statusCode = response.getStatusLine().getStatusCode();

        if (statusCode == 200) {
            return true;
        } else if (statusCode == 404) {
            return false;
        } else {
            logger.warn("Problem while checking index existence: {}", response.getStatusLine().getReasonPhrase());
            throw new QueryExecutionException("Could not check index existence, status code is " + statusCode);
        }
    } catch (IOException e) {
        logger.warn("Problem while verifying if index exists.", e);
        throw new IndexApiException("Error when checking for existing index.");
    }
}

public void dropIndex(String indexName) {
    try {
        Response response = client.performRequest(
                "DELETE",
                indexName);

        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode > 499) {
            logger.warn("Problem while deleting an index: {}", response.getStatusLine().getReasonPhrase());
            throw new QueryExecutionException("Could not delete index, status code is " + statusCode);
        }
    } catch (IOException e) {
        logger.warn("Problem deleting index.");
        throw new IndexApiException("Problem deleting index", e);
    }

}

/**
 * Removes the document with the provided unique identification.
 *
 * @param request Request object containing the required parameters
 * @return Message line that can be used to see if we succeeded.
 */
public String remove(DeleteRequest request) {
    try {
        String endpoint = createEndpointString(request.getIndex(), request.getType(), request.getId());
        if (!request.isMustExist() &&
                !exists(new ExistsRequest(request.getIndex(), request.getType(), request.getId()))) {
            return "not_exists";
        }
        Response response = client.performRequest(DELETE, endpoint);

        return response.getStatusLine().getReasonPhrase();
    } catch (IOException e) {
        logger.warn("Problem while removing a document.", e);
        throw new IndexDocumentException("Error when removing a document");
    }
}