Elasticsearch bulk API returning 400 error in Java

1.7k views Asked by At

I tried a lot of things to check why is it showing a malformed request. Refer to this question to properly format my request body.

Parent specification in elasticsearch for first node

Here is my code.

I am using Unirest API to send requests.

In the commented part of the code the request body is stored. When using the file content and sending a request independently using POSTman, I am able to achieve my target.

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Writer;
import java.util.ArrayList;

import org.json.JSONArray;
import org.json.JSONObject;
import org.jsoup.Jsoup;

import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.JsonNode;
import com.mashape.unirest.http.Unirest;

public class Indexing {

    private static ArrayList<Integer> waitingQueue = new ArrayList<Integer>();
    private static ArrayList<Integer> processedQueue = new ArrayList<Integer>();

    private static String baseURL = "{someURL}";
    private static String appendedURL = "someExtraURL";
    private static Integer initialId = 0000;

    private static String elastiSearchURL = "http://localhost:";
    private static int elastiSearchPort = 9200;
    private static String elasticSeachIndex = "/{someIndex}";
    private static String elasticSearchIndexType = "/{someType}";

    private static ArrayList<String> finalWriteValue = new ArrayList<String>();

    public static void main(String[] args) {
        long startTime = System.currentTimeMillis();

        boolean flag = true;

        waitingQueue.add(initialId);

        while (flag) {
            flag = startCrawling();
        }

        /* File Operation Part
        File file = new File("finalOutput.txt");
        Writer fileWriter = null;
        BufferedWriter bufferedWriter = null;

        try {

            fileWriter = new FileWriter(file);
            bufferedWriter = new BufferedWriter(fileWriter);

            // Write the lines one by one
            for (String line : finalWriteValue) {
                bufferedWriter.write(line);
                bufferedWriter.newLine();
                // alternatively add bufferedWriter.newLine() to change line
            }

        } catch (IOException e) {
            System.err.println("Error writing the file : ");
            e.printStackTrace();
        } finally {

            if (bufferedWriter != null && fileWriter != null) {
                try {
                    bufferedWriter.close();
                    fileWriter.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }*/

        waitingQueue.clear();
        processedQueue.clear();

        StringBuffer body = new StringBuffer();

        for (int i = 0; i < finalWriteValue.size() - 1; i++) {
            body.append(finalWriteValue.get(i));
            body.append("\n");
        }

        body.append(finalWriteValue.get(finalWriteValue.size() - 1));

        // System.out.println(body);
        // System.out.println(body);

        HttpResponse<String> response = null;
        try {
            response = Unirest
                    .post(elastiSearchURL + elastiSearchPort
                            + elasticSeachIndex + elasticSearchIndexType
                            + "/_bulk").body(body.toString()).asString();
        } catch (UnirestException e) { // TODO Auto-generated catch block
            e.printStackTrace();
        }

        System.out.println(response.getStatus());
        finalWriteValue.clear();


        long endTime = System.currentTimeMillis();
        long totalTime = endTime - startTime;
        System.out.println(totalTime);
    }

    private static HttpResponse<JsonNode> apiCall(Integer id) {

        HttpResponse<JsonNode> response = null;

        // System.out.println(baseURL+id.toString()+appendedURL);

        try {
            response = Unirest.get(baseURL + id.toString() + appendedURL)
                    .header("authorization", "{someAuth}")
                    .header("accept", "application/json").asJson();

        } catch (Exception e) {
            System.out.println("------" + e);
        }

        return response;
    }

    protected static String htmlToTextConversion(String htmlText) {

        return Jsoup.parse(htmlText).text();
    }

    private static boolean startCrawling() {
        // TODO Auto-generated method stub

        JSONObject jsonResponse = null;
        JSONObject children = null;
        JSONObject body = null;
        JSONArray content = null;
        JSONObject preText = new JSONObject();
        JSONObject idJSON = new JSONObject();

        if (!waitingQueue.isEmpty()) {
            int position = 0; // waitingQueue.size() - 1;
            int id = waitingQueue.get(position);
            HttpResponse<JsonNode> response = apiCall(id);

            if (response != null) {

                jsonResponse = response.getBody().getObject();
                children = (JSONObject) jsonResponse.get("children");
                body = (JSONObject) jsonResponse.get("body");

                idJSON.put("_id", String.valueOf(id));

                if (jsonResponse.has("parentId"))
                    idJSON.put("parent", jsonResponse.get("parentId"));

                preText.put("index", idJSON);

                String refinedValue = htmlToTextConversion(body
                        .getString("value"));

                body.put("value", refinedValue);

                jsonResponse.put("body", body);

                System.out.println(jsonResponse);
                finalWriteValue.add(preText.toString());
                finalWriteValue.add(jsonResponse.toString());

                content = children.getJSONArray("content");

                for (int i = 0, size = content.length(); i < size; i++) {
                    JSONObject jsonObj = content.getJSONObject(i);

                    Integer tempID = Integer.valueOf(jsonObj.getString("id"));

                    if (waitingQueue.contains(tempID)
                            || processedQueue.contains(tempID)) {
                        continue;
                    } else {
                        waitingQueue.add(tempID);
                    }
                }

                processedQueue.add(waitingQueue.get(position));
                waitingQueue.remove(position);

            } else {
                System.out.println("No response for Page ID "
                        + waitingQueue.get(position));
            }

            return true;
        } else {

            return false;
        }

    }
}

ES Log :

[2016-03-16 10:15:05,666][INFO ][rest.suppressed          ] /helloworld/hello/7639575 Params: {index=helloworld, id=7639575, type=hello} [hello] IndexNotFoundException[no such index]
at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.concreteIndices(IndexNameExpressionResolver.java:151)
at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.concreteIndices(IndexNameExpressionResolver.java:95)
at org.elasticsearch.cluster.metadata.IndexNameExpressionResolver.concreteSingleIndex(IndexNameExpressionResolver.java:208)
at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$AsyncSingleAction.<init>(TransportSingleShardAction.java:138)
at org.elasticsearch.action.support.single.shard.TransportSingleShardAction$AsyncSingleAction.<init>(TransportSingleShardAction.java:115)
at org.elasticsearch.action.support.single.shard.TransportSingleShardAction.doExecute(TransportSingleShardAction.java:87)
at org.elasticsearch.action.support.single.shard.TransportSingleShardAction.doExecute(TransportSingleShardAction.java:51)
at org.elasticsearch.action.support.TransportAction.execute(TransportAction.java:70)
at org.elasticsearch.client.node.NodeClient.doExecute(NodeClient.java:58)
at org.elasticsearch.client.support.AbstractClient.execute(AbstractClient.java:351)
at org.elasticsearch.client.FilterClient.doExecute(FilterClient.java:52)
at org.elasticsearch.rest.BaseRestHandler$HeadersAndContextCopyClient.doExecute(BaseRestHandler.java:83)
at org.elasticsearch.client.support.AbstractClient.execute(AbstractClient.java:351)
at org.elasticsearch.client.support.AbstractClient.get(AbstractClient.java:443)
at org.elasticsearch.rest.action.get.RestGetAction.handleRequest(RestGetAction.java:74)
at org.elasticsearch.rest.BaseRestHandler.handleRequest(BaseRestHandler.java:54)
at org.elasticsearch.rest.RestController.executeHandler(RestController.java:207)
at org.elasticsearch.rest.RestController.dispatchRequest(RestController.java:166)
at org.elasticsearch.http.HttpServer.internalDispatchRequest(HttpServer.java:128)
at org.elasticsearch.http.HttpServer$Dispatcher.dispatchRequest(HttpServer.java:86)
at org.elasticsearch.http.netty.NettyHttpServerTransport.dispatchRequest(NettyHttpServerTransport.java:363)
at org.elasticsearch.http.netty.HttpRequestHandler.messageReceived(HttpRequestHandler.java:63)
at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.elasticsearch.http.netty.pipelining.HttpPipeliningHandler.messageReceived(HttpPipeliningHandler.java:60)
at org.jboss.netty.channel.SimpleChannelHandler.handleUpstream(SimpleChannelHandler.java:88)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.jboss.netty.handler.codec.http.HttpChunkAggregator.messageReceived(HttpChunkAggregator.java:145)
at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.jboss.netty.handler.codec.http.HttpContentDecoder.messageReceived(HttpContentDecoder.java:108)
at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296)
at org.jboss.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:459)
at org.jboss.netty.handler.codec.replay.ReplayingDecoder.callDecode(ReplayingDecoder.java:536)
at org.jboss.netty.handler.codec.replay.ReplayingDecoder.messageReceived(ReplayingDecoder.java:435)
at org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791)
at org.elasticsearch.common.netty.OpenChannelsHandler.handleUpstream(OpenChannelsHandler.java:75)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564)
at org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:559)
at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268)
at org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255)
at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88)
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:108)
at org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:337)
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89)
at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178)
at org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108)
at org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
0

There are 0 answers