Fix for [BUG] Data Prepper is losing connections from S3 pool

Signed-off-by: Krishna Kondaka <[email protected]>
kkondaka · Dec 8, 2023 · 95ce39b · 95ce39b
1 parent 668066f
commit 95ce39b
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 32 deletions.
diff --git a/data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/InputCodec.java b/data-prepper-api/src/main/java/org/opensearch/dataprepper/model/codec/InputCodec.java
@@ -11,6 +11,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.function.Consumer;
+import java.util.Objects;
 
 public interface InputCodec {
     /**
@@ -31,8 +32,19 @@ public interface InputCodec {
      * @param eventConsumer The consumer which handles each event from the stream
      * @throws IOException throws IOException when invalid input is received or incorrect codec name is provided
      */
-    void parse(
+    default void parse(
             InputFile inputFile,
             DecompressionEngine decompressionEngine,
-            Consumer<Record<Event>> eventConsumer) throws IOException;
+            Consumer<Record<Event>> eventConsumer) throws IOException {
+        Objects.requireNonNull(inputFile);
+        Objects.requireNonNull(eventConsumer);
+        InputStream inputStream = inputFile.newStream();
+        try {
+            parse(decompressionEngine.createInputStream(inputStream), eventConsumer);
+            inputStream.close();
+        } catch (Exception e) {
+            inputStream.close();
+            throw e;
+        }
+    }
 }
diff --git a/...ro-codecs/src/main/java/org/opensearch/dataprepper/plugins/codec/avro/AvroInputCodec.java b/...ro-codecs/src/main/java/org/opensearch/dataprepper/plugins/codec/avro/AvroInputCodec.java
@@ -47,14 +47,6 @@ public void parse(InputStream inputStream, Consumer<Record<Event>> eventConsumer
         parseAvroStream(inputStream, eventConsumer);
     }
 
-    @Override
-    public void parse(final InputFile inputFile, final DecompressionEngine decompressionEngine, final Consumer<Record<Event>> eventConsumer) throws IOException {
-        Objects.requireNonNull(inputFile);
-        Objects.requireNonNull(eventConsumer);
-
-        parse(decompressionEngine.createInputStream(inputFile.newStream()), eventConsumer);
-    }
-
     private void parseAvroStream(final InputStream inputStream, final Consumer<Record<Event>> eventConsumer) {
 
         try {
@@ -107,4 +99,4 @@ else if(value instanceof Utf8){
         return eventData;
     }
 
-}
+}
diff --git a/...v-processor/src/main/java/org/opensearch/dataprepper/plugins/codec/csv/CsvInputCodec.java b/...v-processor/src/main/java/org/opensearch/dataprepper/plugins/codec/csv/CsvInputCodec.java
@@ -52,11 +52,6 @@ public void parse(final InputStream inputStream, final Consumer<Record<Event>> e
         }
     }
 
-    @Override
-    public void parse(final InputFile inputFile, DecompressionEngine decompressionEngine, Consumer<Record<Event>> eventConsumer) throws IOException {
-        parse(decompressionEngine.createInputStream(inputFile.newStream()), eventConsumer);
-    }
-
     private void parseBufferedReader(final BufferedReader reader, final Consumer<Record<Event>> eventConsumer) throws IOException {
         final CsvMapper mapper = createCsvMapper();
         final CsvSchema schema;

diff --git a/...ain/java/org/opensearch/dataprepper/plugins/codec/newline/NewlineDelimitedInputCodec.java b/...ain/java/org/opensearch/dataprepper/plugins/codec/newline/NewlineDelimitedInputCodec.java
@@ -48,11 +48,6 @@ public void parse(final InputStream inputStream, final Consumer<Record<Event>> e
         }
     }
 
-    @Override
-    public void parse(final InputFile inputFile, final DecompressionEngine decompressionEngine, final Consumer<Record<Event>> eventConsumer) throws IOException {
-        parse(decompressionEngine.createInputStream(inputFile.newStream()), eventConsumer);
-    }
-
     private void parseBufferedReader(final BufferedReader reader, final Consumer<Record<Event>> eventConsumer) throws IOException {
         final boolean doAddHeaderToOutgoingEvents = Objects.nonNull(headerDestination);
         boolean hasReadHeader = false;

diff --git a/...processor/src/main/java/org/opensearch/dataprepper/plugins/codec/json/JsonInputCodec.java b/...processor/src/main/java/org/opensearch/dataprepper/plugins/codec/json/JsonInputCodec.java
@@ -14,6 +14,7 @@
 import org.opensearch.dataprepper.model.codec.JsonDecoder;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Objects;
 import java.util.function.Consumer;
 
@@ -23,15 +24,4 @@
 @DataPrepperPlugin(name = "json", pluginType = InputCodec.class)
 public class JsonInputCodec extends JsonDecoder implements InputCodec {
 
-    @Override
-    public void parse(
-            final InputFile inputFile,
-            final DecompressionEngine decompressionEngine,
-            final Consumer<Record<Event>> eventConsumer) throws IOException {
-        Objects.requireNonNull(inputFile);
-        Objects.requireNonNull(eventConsumer);
-
-        parse(decompressionEngine.createInputStream(inputFile.newStream()), eventConsumer);
-    }
-
 }