Move RefCountingRunnable, FailProcessorException and Releasable usages to logstash-bridge.

Author: mashhurs

src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessor.java

@@ -13,9 +13,9 @@
 import com.google.common.collect.Maps;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.elasticsearch.action.support.RefCountingRunnable;
-import org.elasticsearch.ingest.common.FailProcessorException;
+import org.elasticsearch.logstashbridge.core.FailProcessorExceptionBridge;
 import org.elasticsearch.logstashbridge.core.IOUtilsBridge;
+import org.elasticsearch.logstashbridge.core.RefCountingRunnableBridge;
 import org.elasticsearch.logstashbridge.ingest.IngestDocumentBridge;
 
 import java.io.Closeable;
@@ -93,8 +93,11 @@ public Collection<Event> processEvents(final Collection<Event> incomingEvents) t
         final CountDownLatch latch = new CountDownLatch(1);
         final IntegrationBatch batch = new IntegrationBatch(incomingEvents);
 
-        try (RefCountingRunnable ref = new RefCountingRunnable(latch::countDown)) {
+        RefCountingRunnableBridge ref = new RefCountingRunnableBridge(latch::countDown);
+        try {
             batch.eachRequest(ref::acquire, this::processRequest);
+        } finally {
+            ref.close();
         }
 
         // await on work that has gone async
@@ -254,7 +257,9 @@ static private void annotateIngestPipelineFailure(final Event event, final Strin
     }
 
     static private Throwable unwrapException(final Exception exception) {
-        if (exception.getCause() instanceof FailProcessorException) { return exception.getCause(); }
+        if (FailProcessorExceptionBridge.isInstanceOf(exception.getCause())) {
+            return exception.getCause();
+        }
         return exception;
     }
 

src/main/java/co/elastic/logstash/filters/elasticintegration/IntegrationBatch.java

@@ -7,7 +7,7 @@
 package co.elastic.logstash.filters.elasticintegration;
 
 import co.elastic.logstash.api.Event;
-import org.elasticsearch.core.Releasable;
+import org.elasticsearch.logstashbridge.core.ReleasableBridge;
 
 import java.util.ArrayList;
 import java.util.Collection;
@@ -22,17 +22,17 @@ public IntegrationBatch(Collection<Event> events) {
         this.events = new ArrayList<>(events);
     }
 
-    void eachRequest(Supplier<Releasable> releasableSupplier, Consumer<IntegrationRequest> consumer) {
+    void eachRequest(Supplier<ReleasableBridge> releasableSupplier, Consumer<IntegrationRequest> consumer) {
         for (int i = 0; i < this.events.size(); i++) {
             consumer.accept(new Request(i, releasableSupplier.get()));
         }
     }
 
     private class Request implements IntegrationRequest {
         private final int idx;
-        private final Releasable handle;
+        private final ReleasableBridge handle;
 
-        public Request(final int idx, final Releasable releasable) {
+        public Request(final int idx, final ReleasableBridge releasable) {
             this.idx = idx;
             this.handle = releasable;
         }