package com.github.ingarabr.mi;

import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;

import org.elasticsearch.client.Client;

public class EsWriter implements Runnable {

    private final Client esClient;
    private final Queue<String> writeQueue = new LinkedBlockingQueue<String>();

    private boolean run = true;

    public EsWriter(Client esClient) {
        this.esClient = esClient;
    }

    public void run() {
        while (run) {
            String metric = writeQueue.poll();
            log();
            esClient.prepareIndex("metrics", "metric").setSource(metric).get();
        }
    }

    private void log() {
        if (writeQueue.size() > 10) {
            System.out.println("ElasticSearch WriteQueue size: " + writeQueue.size());
        }
    }

    public void addMetric(String metric) {
        writeQueue.add(metric);
    }

    public void stop() {
        run = false;
    }
}

@@ -33,26 +33,85 @@
 %0Aimport 
-java.util.
+org.elasticsearch.client.Client;%0Aimport java.util.concurrent.Blocking
 Queue;%0Ai
@@ -163,49 +163,8 @@
 e;%0A%0A
-import org.elasticsearch.client.Client;%0A%0A
 publ
@@ -257,16 +257,24 @@
 e final 
+Blocking
 Queue%3CSt
@@ -461,24 +461,42 @@
 oid run() %7B%0A
+        try %7B%0A    
         whil
@@ -501,24 +501,28 @@
 ile (run) %7B%0A
+    
             
@@ -552,16 +552,20 @@
 eue.
-poll();%0A
+take();%0A    
     
@@ -575,24 +575,70 @@
      log();%0A
+                if (metric != null) %7B%0A        
             
@@ -701,24 +701,121 @@
 ric).get();%0A
+                %7D%0A            %7D%0A        %7D catch (Exception e) %7B%0A            e.printStackTrace();%0A
         %7D%0A  

Use blocking call to queue and null safe it.