handle null records case

amazon-archives · Feb 13, 2017 · af27613 · af27613
1 parent 368e526
commit af27613
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 0 deletions.
diff --git a/src/main/java/com/amazonaws/services/kinesis/connectors/firehose/RecordBatcher.java b/src/main/java/com/amazonaws/services/kinesis/connectors/firehose/RecordBatcher.java
@@ -31,6 +31,10 @@ List<List<Record>> makeBatches(List<Record> records) {
 
         for (Record record : records) {
 
+            if (record == null) {
+                continue;
+            }
+
             Integer recSize = record.getData().remaining();
 
             if (numRecords + 1 > 500 || byteSize + recSize > 4194304) {

diff --git a/src/test/java/com/amazonaws/services/kinesis/connectors/firehose/FirehoseEmitterTest.java b/src/test/java/com/amazonaws/services/kinesis/connectors/firehose/FirehoseEmitterTest.java
@@ -97,6 +97,20 @@ public void testNoRecordsCase() throws IOException {
         verify(firehoseClientMock, buffer);
     }
 
+    @Test
+    public void testNullRecordsCase() throws IOException {
+        ArrayList<Record> recs = new ArrayList<Record>();
+        recs.add(null);
+        expect(buffer.getRecords()).andReturn(recs);
+
+        replay(firehoseClientMock, buffer);
+
+        List<Record> failures = emitter.emit(buffer);
+        Assert.assertTrue(failures.isEmpty());
+
+        verify(firehoseClientMock, buffer);
+    }
+
     @Test
     public void testWorkingCaseNoFailures() throws IOException, InterruptedException, ExecutionException {
         List<Record> records = generateRecords(1750);