Skip to content

NIFI-12745: Fix AvroReader silently dropping malformed records #8361

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
takraj wants to merge 4 commits into from
Closed

NIFI-12745: Fix AvroReader silently dropping malformed records #8361

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
d2c7abd
NIFI-12745: Fix AvroReader silently dropping malformed records
takraj Feb 6, 2024
c334b98
Apply suggestions from code review
takraj Feb 8, 2024
dd74edd
Fix imports
takraj Feb 8, 2024
6c7751c
Update nifi-nar-bundles/nifi-standard-services/nifi-record-serializat…
takraj Feb 8, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions ...ices/nifi-record-serialization-services-bundle/nifi-record-serialization-services/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,8 @@
<exclude>src/test/resources/avro/multiple-types.avsc</exclude>
<exclude>src/test/resources/avro/simple.avsc</exclude>
<exclude>src/test/resources/avro/recursive.avsc</exclude>
<exclude>src/test/resources/avro/schemaless_simple_record.avsc</exclude>
<exclude>src/test/resources/avro/schemaless_simple_record_extra_field.avsc</exclude>

<exclude>src/test/resources/cef/empty-row.txt</exclude>
<exclude>src/test/resources/cef/misformatted-row.txt</exclude>
Expand Down
2 changes: 1 addition & 1 deletion ...rialization-services/src/main/java/org/apache/nifi/avro/AvroReaderWithExplicitSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ protected GenericRecord nextAvroRecord() throws IOException {
try {
genericRecord = datumReader.read(null, decoder);
} catch (final EOFException eof) {
return null;
throw new IOException("Was expecting more data, but reached EOF.", eof);
}

return genericRecord;
Expand Down
54 changes: 41 additions & 13 deletions ...ization-services/src/test/java/org/apache/nifi/avro/TestAvroReaderWithExplicitSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,10 @@
*/
package org.apache.nifi.avro;

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordSchema;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;

import java.io.File;
import java.io.FileInputStream;
Expand All @@ -32,11 +28,15 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.nifi.serialization.MalformedRecordException;
import org.apache.nifi.serialization.RecordReader;
import org.apache.nifi.serialization.SimpleRecordSchema;
import org.apache.nifi.serialization.record.Record;
import org.apache.nifi.serialization.record.RecordSchema;
import org.junit.jupiter.api.Test;

public class TestAvroReaderWithExplicitSchema {

Expand Down Expand Up @@ -105,6 +105,34 @@ public void testAvroExplicitReaderWithEmbeddedSchemaFileDifferentFromExplicitSch
assertThrows(IOException.class, () -> new AvroReaderWithExplicitSchema(fileInputStream, recordSchema, dataSchema));
}

@Test
public void testAvroExplicitReaderWithSchemalessFileAndExplicitSchema() throws Exception {
AvroReaderWithExplicitSchema avroReader = createAvroReaderWithExplicitSchema(
"src/test/resources/avro/schemaless_simple_record.avro",
"src/test/resources/avro/schemaless_simple_record.avsc"
);

GenericData.Record expected = new GenericData.Record(new Schema.Parser().parse(new File("src/test/resources/avro/schemaless_simple_record.avsc")));
expected.put("field_1", 123);
expected.put("field_2", "44");
expected.put("field_3", 5);

GenericRecord actual1 = avroReader.nextAvroRecord();
assertEquals(expected, actual1);

GenericRecord actual2 = avroReader.nextAvroRecord();
assertNull(actual2);
}

@Test
public void testAvroExplicitReaderWithSchemalessFileAndWrongExplicitSchema() throws Exception {
AvroReaderWithExplicitSchema avroReader = createAvroReaderWithExplicitSchema(
"src/test/resources/avro/schemaless_simple_record.avro",
"src/test/resources/avro/schemaless_simple_record_extra_field.avsc"
);
assertThrows(IOException.class, avroReader::nextAvroRecord);
}

@Test
public void testAvroExplicitReaderWithSchemalessFileDecimalValuesWithDifferentBufferSize() throws Exception {
// GIVEN
Expand Down
Binary file added .../nifi-record-serialization-services/src/test/resources/avro/schemaless_simple_record.avro
View file
Open in desktop
Binary file not shown.
28 changes: 28 additions & 0 deletions .../nifi-record-serialization-services/src/test/resources/avro/schemaless_simple_record.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
{
"type": "record",
"name": "nifiRecord",
"namespace": "org.apache.nifi",
"fields": [
{
"name": "field_1",
"type": [
"int",
"null"
]
},
{
"name": "field_2",
"type": [
"string",
"null"
]
},
{
"name": "field_3",
"type": [
"int",
"null"
]
}
]
}
24 changes: 24 additions & 0 deletions ...-serialization-services/src/test/resources/avro/schemaless_simple_record_extra_field.avsc
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,24 @@
{
"type":"record",
"name":"message_name",
"namespace":"message_namespace",
"fields":[
{
"name":"field_1",
"type":["long"]
},
{
"name":"field_2",
"type":["string"]
},
{
"name":"field_3",
"type":["int"]
},
{
"name":"extra",
"type":["long"]
}
]
}