package org.sanju.kafka.connect.marklogic;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;

import org.apache.http.HttpResponse;
import org.apache.kafka.connect.errors.RetriableException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @author Sanju Thomas
 *
 */
public class MarkLogicAsyncWriter extends MarkLogicWriter {

    private static final Logger logger = LoggerFactory.getLogger(MarkLogicAsyncWriter.class);

    public MarkLogicAsyncWriter(final Map<String, String> config) {
        super(config);
    }

    @Override
    public void write(final Collection<SinkRecord> rs) {

        final ExecutorService fjPool = new ForkJoinPool(8);
        final List<Callable<HttpResponse>> rps = new ArrayList<>();
        try {
            rs.forEach(r -> {
                rps.add(new RequestProcessor(r));
            });
            final List<Future<HttpResponse>> results = fjPool.invokeAll(rps);
            results.forEach(r -> {
                try {
                    r.get();
                } catch (Exception e) {
                    throw new RetriableException(e);
                }
            });
        } catch (Exception e) {
            logger.error("Exception occurred during the write, marking the entire collection of records with size {} to retry, exception was {}",
                    rs.size(), e);
            throw new RetriableException(e);
        }finally{
            fjPool.shutdown();
        }
    }

    class RequestProcessor implements Callable<HttpResponse> {

        private SinkRecord record;

        public RequestProcessor(final SinkRecord record) {
            this.record = record;
        }

        @Override
        public HttpResponse call() {
            return process(createPutRequest(record.value(), record.topic()));
        }
    }
}

@@ -1057,43 +1057,122 @@
     
-    rps.add(new RequestProcessor(r)
+%09final Callable%3CHttpResponse%3E c = () -%3E process(createPutRequest(r.value(), r.topic()));%0A                rps.add(c
 );%0A 
@@ -1805,359 +1805,6 @@
  %7D%0A%0A
-    class RequestProcessor implements Callable%3CHttpResponse%3E %7B%0A%0A        private SinkRecord record;%0A%0A        public RequestProcessor(final SinkRecord record) %7B%0A            this.record = record;%0A        %7D%0A%0A        @Override%0A        public HttpResponse call() %7B%0A            return process(createPutRequest(record.value(), record.topic()));%0A        %7D%0A    %7D%0A
 %7D%0A

Update MarkLogicAsyncWriter.java