You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ex...@apache.org on 2024/11/12 14:50:17 UTC
(nifi) branch support/nifi-1.x updated: NIFI-13963 Default to Drop Unknown Fields in JSON Record Reader (#9485)
This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch support/nifi-1.x
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/support/nifi-1.x by this push:
new 9969eca817 NIFI-13963 Default to Drop Unknown Fields in JSON Record Reader (#9485)
9969eca817 is described below
commit 9969eca817e8f1b7311ec42a5e9f1e30a708494f
Author: dan-s1 <ds...@gmail.com>
AuthorDate: Tue Nov 12 09:50:10 2024 -0500
NIFI-13963 Default to Drop Unknown Fields in JSON Record Reader (#9485)
- Ensured that unknown fields are not included in output schema for JSON Records
Signed-off-by: David Handermann <ex...@apache.org>
---
.../nifi/serialization/record/MapRecord.java | 3 --
.../nifi/json/AbstractJsonRowRecordReader.java | 4 +++
.../apache/nifi/json/JsonTreeRowRecordReader.java | 17 ++++++++++-
.../nifi-standard-processors/pom.xml | 1 +
.../processors/standard/TestConvertRecord.java | 34 ++++++++++++++++++++++
.../TestConvertRecord/input/person_dropfield.json | 9 ++++++
6 files changed, 64 insertions(+), 4 deletions(-)
diff --git a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java
index a3c5dd522d..059d6c8d91 100644
--- a/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java
+++ b/nifi-commons/nifi-record/src/main/java/org/apache/nifi/serialization/record/MapRecord.java
@@ -514,9 +514,6 @@ public class MapRecord implements Record {
private Optional<RecordField> setValueAndGetField(final String fieldName, final Object value) {
final Optional<RecordField> field = getSchema().getField(fieldName);
if (!field.isPresent()) {
- if (dropUnknownFields) {
- return field;
- }
final Object previousValue = values.put(fieldName, value);
if (!Objects.equals(value, previousValue)) {
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
index 0fef6d50c1..ba1ddab873 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/AbstractJsonRowRecordReader.java
@@ -185,6 +185,10 @@ public abstract class AbstractJsonRowRecordReader implements RecordReader {
return lazyTimestampFormat;
}
+ @Override
+ public Record nextRecord() throws IOException, MalformedRecordException {
+ return nextRecord(true, true);
+ }
@Override
public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException {
diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
index 489474e804..8655ec0386 100644
--- a/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
+++ b/nifi-nar-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/JsonTreeRowRecordReader.java
@@ -147,8 +147,22 @@ public class JsonTreeRowRecordReader extends AbstractJsonRowRecordReader {
final boolean coerceTypes, final boolean dropUnknown) throws IOException, MalformedRecordException {
final Map<String, Object> values = new LinkedHashMap<>(schema.getFieldCount() * 2);
+ final JsonNode jsonNodeForSerialization;
if (dropUnknown) {
+ jsonNodeForSerialization = jsonNode.deepCopy();
+
+ // Delete unknown fields for updated serialized representation
+ final Iterator<Map.Entry<String, JsonNode>> fields = jsonNodeForSerialization.fields();
+ while (fields.hasNext()) {
+ final Map.Entry<String, JsonNode> field = fields.next();
+ final String fieldName = field.getKey();
+ final Optional<RecordField> recordField = schema.getField(fieldName);
+ if (!recordField.isPresent()) {
+ fields.remove();
+ }
+ }
+
for (final RecordField recordField : schema.getFields()) {
final JsonNode childNode = getChildNode(jsonNode, recordField);
if (childNode == null) {
@@ -169,6 +183,7 @@ public class JsonTreeRowRecordReader extends AbstractJsonRowRecordReader {
values.put(fieldName, value);
}
} else {
+ jsonNodeForSerialization = jsonNode;
final Iterator<String> fieldNames = jsonNode.fieldNames();
while (fieldNames.hasNext()) {
final String fieldName = fieldNames.next();
@@ -189,7 +204,7 @@ public class JsonTreeRowRecordReader extends AbstractJsonRowRecordReader {
}
}
- final Supplier<String> supplier = jsonNode::toString;
+ final Supplier<String> supplier = jsonNodeForSerialization::toString;
return new MapRecord(schema, values, SerializedForm.of(supplier, "application/json"), false, dropUnknown);
}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
index 231df0e370..43147a765f 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml
@@ -557,6 +557,7 @@
<exclude>src/test/resources/TestConvertRecord/input/person_bad_enum.json</exclude>
<exclude>src/test/resources/TestConvertRecord/input/person_long_id.json</exclude>
<exclude>src/test/resources/TestConvertRecord/input/person.json</exclude>
+ <exclude>src/test/resources/TestConvertRecord/input/person_dropfield.json</exclude>
<exclude>src/test/resources/TestConvertRecord/schema/person_with_union_enum_string.avsc</exclude>
<exclude>src/test/resources/TestConvertRecord/schema/person.avsc</exclude>
<exclude>src/test/resources/TestCountText/jabberwocky.txt</exclude>
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java
index 189caecc98..a99f0943fd 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestConvertRecord.java
@@ -51,6 +51,7 @@ import java.util.HashMap;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@DisabledOnOs(value = OS.WINDOWS, disabledReason = "Pretty-printing is not portable across operating systems")
@@ -404,4 +405,37 @@ public class TestConvertRecord {
}
}
}
+
+ @Test
+ public void testJSONDroppingUnkownFields() throws InitializationException, IOException {
+ final TestRunner runner = TestRunners.newTestRunner(ConvertRecord.class);
+ final JsonTreeReader jsonReader = new JsonTreeReader();
+ runner.addControllerService("reader", jsonReader);
+
+ final String inputSchemaText = new String(Files.readAllBytes(Paths.get("src/test/resources/TestConvertRecord/schema/person.avsc")));
+ final String outputSchemaText = new String(Files.readAllBytes(Paths.get("src/test/resources/TestConvertRecord/schema/person.avsc")));
+
+ runner.setProperty(jsonReader, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY, SchemaAccessUtils.SCHEMA_TEXT_PROPERTY);
+ runner.setProperty(jsonReader, SchemaAccessUtils.SCHEMA_TEXT, inputSchemaText);
+ runner.enableControllerService(jsonReader);
+
+ final JsonRecordSetWriter jsonWriter = new JsonRecordSetWriter();
+ runner.addControllerService("writer", jsonWriter);
+ runner.setProperty(jsonWriter, SchemaAccessUtils.SCHEMA_ACCESS_STRATEGY, SchemaAccessUtils.SCHEMA_TEXT_PROPERTY);
+ runner.setProperty(jsonWriter, SchemaAccessUtils.SCHEMA_TEXT, outputSchemaText);
+ runner.setProperty(jsonWriter, "Schema Write Strategy", "full-schema-attribute");
+ runner.enableControllerService(jsonWriter);
+
+ runner.enqueue(Paths.get("src/test/resources/TestConvertRecord/input/person_dropfield.json"));
+
+ runner.setProperty(ConvertRecord.RECORD_READER, "reader");
+ runner.setProperty(ConvertRecord.RECORD_WRITER, "writer");
+
+ runner.run();
+ runner.assertAllFlowFilesTransferred(ConvertRecord.REL_SUCCESS, 1);
+
+ MockFlowFile flowFile = runner.getFlowFilesForRelationship(ConvertRecord.REL_SUCCESS).get(0);
+ final String content = flowFile.getContent();
+ assertFalse(content.contains("fieldThatShouldBeRemoved"));
+ }
}
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestConvertRecord/input/person_dropfield.json b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestConvertRecord/input/person_dropfield.json
new file mode 100644
index 0000000000..50f955f5d2
--- /dev/null
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestConvertRecord/input/person_dropfield.json
@@ -0,0 +1,9 @@
+[ {
+ "id" : 485,
+ "name" : {
+ "last" : "Doe",
+ "first" : "John"
+ },
+ "status" : "ACTIVE",
+ "fieldThatShouldBeRemoved": "Test"
+} ]
\ No newline at end of file