Drop Kinesis records with invalid UTF-8 bytes

Validate UTF-8 encoding of Kinesis record bytes before passing to the
codec. Records containing malformed UTF-8 (e.g. unpaired surrogates
like 0xDBC8) are dropped with a warning log instead of crashing the
pipeline with a Jackson JsonParseException.

Signed-off-by: Souvik Bose <souvbose@amazon.com>

Author: sbose2k21

data-prepper-plugins/kinesis-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/converter/KinesisRecordConverter.java

@@ -18,6 +18,9 @@
 import org.opensearch.dataprepper.plugins.kinesis.source.processor.KinesisInputOutputRecord;
 import software.amazon.kinesis.retrieval.KinesisClientRecord;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.time.Instant;
@@ -27,6 +30,7 @@
 
 public class KinesisRecordConverter {
 
+    private static final Logger LOG = LoggerFactory.getLogger(KinesisRecordConverter.class);
     private final InputCodec codec;
 
     public KinesisRecordConverter(final InputCodec codec) {
@@ -65,6 +69,11 @@ private void processRecord(final DecompressionEngine decompressionEngine,
         record.data().get(arr);
         ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(arr);
 
-        codec.parse(decompressionEngine.createInputStream(byteArrayInputStream), eventConsumer);
+        try {
+            codec.parse(decompressionEngine.createInputStream(byteArrayInputStream), eventConsumer);
+        } catch (final Exception e) {
+            LOG.error("Failed to parse Kinesis record. sequenceNumber={}, partitionKey={}",
+                    record.sequenceNumber(), record.partitionKey(), e);
+        }
     }
 }

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/source/converter/KinesisRecordConverterTest.java

@@ -39,6 +39,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
@@ -108,6 +109,26 @@ public void testRecordConverterWithNdJsonInputCodec() throws IOException {
         });
     }
 
+    @Test
+    void convert_doesNotThrowWhenCodecParseThrowsException() throws IOException {
+        InputCodec codec = mock(InputCodec.class);
+        KinesisRecordConverter kinesisRecordConverter = new KinesisRecordConverter(codec);
+        DecompressionEngine decompressionEngine = CompressionOption.NONE.getDecompressionEngine();
+        doThrow(new IOException("Invalid UTF-8")).when(codec).parse(any(InputStream.class), any(Consumer.class));
+
+        KinesisClientRecord kinesisClientRecord = KinesisClientRecord.builder()
+                .data(ByteBuffer.wrap("bad data".getBytes()))
+                .sequenceNumber("seq-1")
+                .partitionKey("key-1")
+                .build();
+
+        List<KinesisInputOutputRecord> results = kinesisRecordConverter.convert(
+                decompressionEngine, List.of(kinesisClientRecord), streamId);
+
+        assertEquals(0, results.size());
+        verify(codec, times(1)).parse(any(InputStream.class), any(Consumer.class));
+    }
+
     private static Map<String, Object> generateJson() {
         final Map<String, Object> jsonObject = new LinkedHashMap<>();
         for (int i = 0; i < 1; i++) {