ElasticSearch and Apache HttpAsyncClient

I'm trying to use ElasticSearch REST API with Java Apache HttpAsyncClient library. I want to use persistent pipelining connection. Here is some test code (output is in comments):

@Test
public void testEsPipeliningClient() throws IOException, ExecutionException, InterruptedException
{
    testPost(HttpAsyncClients.createDefault());
    //201: {"_index":"test_index","_type":"test_type","_id":"AVIHYGnqdqqg_TAHm4ix","_version":1,"_shards":{"total":2,"successful":1,"failed":0},"created":true}
    testPost(HttpAsyncClients.createPipelining());
    //400: No handler found for uri [http://127.0.0.1:9200/test_index/test_type] and method [POST]
}

private void testPost(CloseableHttpAsyncClient client) throws ExecutionException, InterruptedException, IOException
{
    client.start();
    HttpPost request = new HttpPost("http://127.0.0.1:9200/test_index/test_type");
    request.setEntity(new StringEntity("{\"some_field\": \"some_value\"}"));
    Future<HttpResponse> responseFuture = client.execute(request, null);
    HttpResponse response = responseFuture.get();
    System.err.println(response.getStatusLine().getStatusCode() + ": " + EntityUtils.toString(response.getEntity()));
}

I can't understand, why it works fine with HttpAsyncClients.createDefault() client, but doesn't work with HttpAsyncClients.createPipelining(). Also I can't understand the difference between these two creation methods.

Why do I get error response when I use createPipelining()?

I tried to see the difference with https://httpbin.org/post but it showed me the same result with both options. I use default ElasticSearch settings.

Thanks!

UPD1

I tried with PUT document (PUT http://127.0.0.1/test_index/test_type/<doc id>) request with the same result - it works fine with createDefault() but I got similar error when do it with createPipelining() - No handler was found <...>.

But when I try to execute request to create index (PUT http://127.0.0.1/<index name>) there is another error. See the code below:

@Test
public void testEsPipeliningClient() throws IOException, ExecutionException, InterruptedException
{
    testCreateIndex(HttpAsyncClients.createDefault());
    //200: {"acknowledged":true}
    testCreateIndex(HttpAsyncClients.createPipelining());
    //400: {"error":{"root_cause":[{"type":"mapper_parsing_exception","reason":"failed to parse, document is empty"}],"type":"mapper_parsing_exception","reason":"failed to parse, document is empty"},"status":400}
}

private void testCreateIndex(CloseableHttpAsyncClient client) throws ExecutionException, InterruptedException, IOException
{
    client.start();
    HttpPut request = new HttpPut("http://127.0.0.1:9200/" + RandomStringUtils.randomAlphabetic(8).toLowerCase());
    Future<HttpResponse> responseFuture = client.execute(request, null);
    HttpResponse response = responseFuture.get();
    System.err.println(response.getStatusLine().getStatusCode() + ": " + EntityUtils.toString(response.getEntity()));
}

As I can see at this documentation page ElasticSearch supports HTTP pipelining by default. Maybe there anything I need to change in ES settings?

UPD2

Here are some wire logs for code in UPD1 section with different logging settings:

Dorg.apache.commons.logging.simplelog.log.org.apache.http=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.wire=INFO

http://pastebin.com/v29uvgbj

-Dorg.apache.commons.logging.simplelog.log.org.apache.http.impl.conn=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.impl.client=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.client=DEBUG -Dorg.apache.commons.logging.simplelog.log.org.apache.http.wire=DEBUG

http://pastebin.com/G9ij15d6

UPD3

I just tried to replace createDefault() with createMinimal() and it caused the same error that createPipelining(). Any ideas what in MinimalHttpAsyncClient may cause this problem? Maybe there is a way I can manually create pipelining client (with builder classes) without this problem?

回答1:

The server must be choking on absolute request URI in the request line

[DEBUG] wire - http-outgoing-1 >> "PUT http://127.0.0.1:9200/ydiwdsid HTTP/1.1[\r][\n]"

HttpAsyncClient in the pipelining mode employs a minimal protocol processing chain. It does not attempt to rewrite the request URI of the request object.

For your particular case request pipelining does not seem to make a lot of sense. Not to mention that unless you are submitting requests in batches you are not even using pipelined execution.

回答2:

Actually, you simply need to extract the host from the URL and create an HttpPost object only with the absolute path. See the changes on the second, third and fifth lines below:

client.start();
HttpHost targetHost = new HttpHost("127.0.0.1", 9200);
HttpPost request = new HttpPost("/test_index/test_type");
request.setEntity(new StringEntity("{\"some_field\": \"some_value\"}"));
Future<HttpResponse> responseFuture = client.execute(targetHost, request, null);
HttpResponse response = responseFuture.get();
System.out.println(response.getStatusLine().getStatusCode() + ": " + EntityUtils.toString(response.getEntity()));

Doing these three changes and running the code again will yield this:

201: {"_index":"test_index","_type":"test_type","_id":"AVISSimIZHOoPG8ibOyF","_version":1,"created":true}
201: {"_index":"test_index","_type":"test_type","_id":"AVISSimjZHOoPG8ibOyG","_version":1,"created":true}