This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new be995d269fc NIFI-15856 Added Serialized JSON Input Handling to
JsonRecordSetWriter (#11158)
be995d269fc is described below
commit be995d269fce9ea63cf39cdc7ecb6e58fe0a1fca
Author: Pierre Villard <[email protected]>
AuthorDate: Thu Apr 23 15:53:39 2026 +0200
NIFI-15856 Added Serialized JSON Input Handling to JsonRecordSetWriter
(#11158)
Signed-off-by: David Handermann <[email protected]>
---
.../java/org/apache/nifi/json/WriteJsonResult.java | 33 +++++-
.../org/apache/nifi/json/JsonRecordSetWriter.java | 25 +++-
.../org/apache/nifi/json/TestWriteJsonResult.java | 127 +++++++++++++++++++++
3 files changed, 183 insertions(+), 2 deletions(-)
diff --git
a/nifi-extension-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/WriteJsonResult.java
b/nifi-extension-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/WriteJsonResult.java
index e3f10a02f71..3aa27abb8d8 100644
---
a/nifi-extension-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/WriteJsonResult.java
+++
b/nifi-extension-bundles/nifi-extension-utils/nifi-record-utils/nifi-json-record-utils/src/main/java/org/apache/nifi/json/WriteJsonResult.java
@@ -67,17 +67,24 @@ public class WriteJsonResult extends
AbstractRecordSetWriter implements RecordSe
private final String mimeType;
private final boolean prettyPrint;
private final boolean allowScientificNotation;
+ private final boolean serializedInputHandlingEnabled;
private static final ObjectMapper objectMapper = new ObjectMapper();
public WriteJsonResult(final ComponentLog logger, final RecordSchema
recordSchema, final SchemaAccessWriter schemaAccess, final OutputStream out,
final boolean prettyPrint,
final NullSuppression nullSuppression, final OutputGrouping
outputGrouping, final String dateFormat, final String timeFormat, final String
timestampFormat) throws IOException {
- this(logger, recordSchema, schemaAccess, out, prettyPrint,
nullSuppression, outputGrouping, dateFormat, timeFormat, timestampFormat,
"application/json", false);
+ this(logger, recordSchema, schemaAccess, out, prettyPrint,
nullSuppression, outputGrouping, dateFormat, timeFormat, timestampFormat,
"application/json", false, true);
}
public WriteJsonResult(final ComponentLog logger, final RecordSchema
recordSchema, final SchemaAccessWriter schemaAccess, final OutputStream out,
final boolean prettyPrint,
final NullSuppression nullSuppression, final OutputGrouping
outputGrouping, final String dateFormat, final String timeFormat, final String
timestampFormat,
final String mimeType, final boolean allowScientificNotation) throws
IOException {
+ this(logger, recordSchema, schemaAccess, out, prettyPrint,
nullSuppression, outputGrouping, dateFormat, timeFormat, timestampFormat,
mimeType, allowScientificNotation, true);
+ }
+
+ public WriteJsonResult(final ComponentLog logger, final RecordSchema
recordSchema, final SchemaAccessWriter schemaAccess, final OutputStream out,
final boolean prettyPrint,
+ final NullSuppression nullSuppression, final OutputGrouping
outputGrouping, final String dateFormat, final String timeFormat, final String
timestampFormat,
+ final String mimeType, final boolean allowScientificNotation, final
boolean serializedInputHandlingEnabled) throws IOException {
super(out);
this.logger = logger;
@@ -87,6 +94,7 @@ public class WriteJsonResult extends AbstractRecordSetWriter
implements RecordSe
this.outputGrouping = outputGrouping;
this.mimeType = mimeType;
this.allowScientificNotation = allowScientificNotation;
+ this.serializedInputHandlingEnabled = serializedInputHandlingEnabled;
this.dateFormat = dateFormat;
this.timeFormat = timeFormat;
@@ -170,7 +178,30 @@ public class WriteJsonResult extends
AbstractRecordSetWriter implements RecordSe
return WriteResult.of(incrementRecordCount(), attributes);
}
+ /**
+ * Determines whether the record's original serialized JSON bytes can be
emitted verbatim as a throughput optimization,
+ * bypassing field-by-field re-serialization. All of the following
conditions must hold for the fast path to apply:
+ * <ol>
+ * <li>The caller enabled the optimization (the {@code
serializedInputHandlingEnabled} constructor argument is {@code true}).</li>
+ * <li>The record carries a {@link SerializedForm} produced by the
upstream reader. Today this is only set by
+ * {@code JsonTreeRowRecordReader}; readers such as {@code
JsonPathRowRecordReader} that transform the input
+ * cannot reuse their input bytes and therefore never trigger the
fast path.</li>
+ * <li>The serialized form's MIME type matches the writer's configured
MIME type and the reader's record schema is
+ * equal to the writer's record schema (no projection, no field
renames, no type coercion).</li>
+ * <li>The cached bytes are a {@code String}.</li>
+ * <li>The cached bytes' pretty-print state matches the writer's {@code
prettyPrint} setting.</li>
+ * <li>If scientific notation is disabled on the writer, the cached
bytes do not contain scientific notation.</li>
+ * </ol>
+ * When the fast path is taken, the writer emits the cached bytes via
{@link JsonGenerator#writeRawValue(String)} and
+ * therefore does <em>not</em> apply the writer's Timestamp Format, Date
Format, Time Format, or Suppress Null Values
+ * settings to that record. Operators that need those writer-side
properties to be honored uniformly must construct
+ * this writer with {@code serializedInputHandlingEnabled = false}.
+ */
private boolean isUseSerializeForm(final Record record, final RecordSchema
writeSchema) {
+ if (!serializedInputHandlingEnabled) {
+ return false;
+ }
+
final Optional<SerializedForm> serializedForm =
record.getSerializedForm();
if (serializedForm.isEmpty()) {
return false;
diff --git
a/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
index 2d5720dd45c..d3062384e45 100644
---
a/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
+++
b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/main/java/org/apache/nifi/json/JsonRecordSetWriter.java
@@ -67,6 +67,15 @@ public class JsonRecordSetWriter extends
DateTimeTextRecordSetWriter implements
public static final AllowableValue OUTPUT_ONELINE = new
AllowableValue("output-oneline", "One Line Per Object",
"Output records with one JSON object per line, delimited by a
newline character");
+ public static final AllowableValue HANDLING_ENABLED = new
AllowableValue("ENABLED", "Enabled",
+ """
+ The writer may emit the input reader's original JSON bytes
verbatim when it can do so safely, as a throughput optimization. \
+ Timestamp Format, Date Format, Time Format, and Suppress
Null Values may not be applied to those records.""");
+ public static final AllowableValue HANDLING_DISABLED = new
AllowableValue("DISABLED", "Disabled",
+ """
+ The writer re-serializes every record from typed field
values, so Timestamp Format, Date Format, Time Format, and Suppress Null \
+ Values are honored uniformly.""");
+
public static final String COMPRESSION_FORMAT_GZIP = "gzip";
public static final String COMPRESSION_FORMAT_BZIP2 = "bzip2";
public static final String COMPRESSION_FORMAT_XZ_LZMA2 = "xz-lzma2";
@@ -123,6 +132,17 @@ public class JsonRecordSetWriter extends
DateTimeTextRecordSetWriter implements
.allowableValues("0", "1", "2", "3", "4", "5", "6", "7", "8", "9")
.dependsOn(COMPRESSION_FORMAT, COMPRESSION_FORMAT_GZIP)
.build();
+ public static final PropertyDescriptor SERIALIZED_JSON_INPUT_HANDLING =
new PropertyDescriptor.Builder()
+ .name("Serialized JSON Input Handling")
+ .description("""
+ When enabled, the writer may emit the input reader's
original JSON bytes verbatim when it can do so safely, as a \
+ throughput optimization. In that case, the Timestamp
Format, Date Format, Time Format, and Suppress Null Values properties may not
be \
+ applied to those records. When disabled, the writer
re-serializes every record so that these properties are honored uniformly.""")
+ .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+ .allowableValues(HANDLING_ENABLED, HANDLING_DISABLED)
+ .defaultValue(HANDLING_ENABLED.getValue())
+ .required(true)
+ .build();
private volatile boolean prettyPrint;
private volatile boolean allowScientificNotation;
@@ -130,6 +150,7 @@ public class JsonRecordSetWriter extends
DateTimeTextRecordSetWriter implements
private volatile OutputGrouping outputGrouping;
private volatile String compressionFormat;
private volatile int compressionLevel;
+ private volatile boolean serializedInputHandlingEnabled;
@Override
protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
@@ -140,6 +161,7 @@ public class JsonRecordSetWriter extends
DateTimeTextRecordSetWriter implements
properties.add(OUTPUT_GROUPING);
properties.add(COMPRESSION_FORMAT);
properties.add(COMPRESSION_LEVEL);
+ properties.add(SERIALIZED_JSON_INPUT_HANDLING);
return properties;
}
@@ -197,6 +219,7 @@ public class JsonRecordSetWriter extends
DateTimeTextRecordSetWriter implements
this.compressionFormat =
context.getProperty(COMPRESSION_FORMAT).getValue();
this.compressionLevel =
context.getProperty(COMPRESSION_LEVEL).asInteger();
+ this.serializedInputHandlingEnabled =
HANDLING_ENABLED.getValue().equals(context.getProperty(SERIALIZED_JSON_INPUT_HANDLING).getValue());
}
@Override
@@ -241,7 +264,7 @@ public class JsonRecordSetWriter extends
DateTimeTextRecordSetWriter implements
}
return new WriteJsonResult(logger, schema,
getSchemaAccessWriter(schema, variables), compressionOut, prettyPrint,
nullSuppression, outputGrouping,
- getDateFormat().orElse(null), getTimeFormat().orElse(null),
getTimestampFormat().orElse(null), mimeType, allowScientificNotation);
+ getDateFormat().orElse(null), getTimeFormat().orElse(null),
getTimestampFormat().orElse(null), mimeType, allowScientificNotation,
serializedInputHandlingEnabled);
}
}
diff --git
a/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java
b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java
index 575286f2dea..35263a8dd70 100644
---
a/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java
+++
b/nifi-extension-bundles/nifi-standard-services/nifi-record-serialization-services-bundle/nifi-record-serialization-services/src/test/java/org/apache/nifi/json/TestWriteJsonResult.java
@@ -52,6 +52,7 @@ import java.util.List;
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;
class TestWriteJsonResult {
@@ -631,4 +632,130 @@ class TestWriteJsonResult {
final String output = new String(data, StandardCharsets.UTF_8);
assertEquals(json, output);
}
+
+ @Test
+ void testReuseInputSerializationDefaultTrueUsesFastPath() throws
IOException {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new RecordField("name",
RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField("age", RecordFieldType.INT.getDataType()));
+ final RecordSchema schema = new SimpleRecordSchema(fields);
+
+ final Map<String, Object> values = new HashMap<>();
+ values.put("name", "John Doe");
+ values.put("age", 42);
+
+ final String rawForm = "{\"name\":\"John
Doe\",\"age\":42,\"ignoredExtra\":\"preserved\"}";
+ final SerializedForm serializedForm = SerializedForm.of(rawForm,
"application/json");
+ final Record record = new MapRecord(schema, values, serializedForm);
+
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final WriteJsonResult writer = new
WriteJsonResult(Mockito.mock(ComponentLog.class), schema, new
SchemaNameAsAttribute(), baos, false,
+ NullSuppression.NEVER_SUPPRESS, OutputGrouping.OUTPUT_ARRAY,
RecordFieldType.DATE.getDefaultFormat(),
+ RecordFieldType.TIME.getDefaultFormat(),
RecordFieldType.TIMESTAMP.getDefaultFormat())) {
+ writer.write(RecordSet.of(schema, record));
+ }
+
+ final String output = baos.toString(StandardCharsets.UTF_8);
+ assertEquals("[{\"name\":\"John
Doe\",\"age\":42,\"ignoredExtra\":\"preserved\"}]", output);
+ }
+
+ @Test
+ void testReuseInputSerializationFalseForcesReserialization() throws
IOException {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new RecordField("name",
RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField("age", RecordFieldType.INT.getDataType()));
+ final RecordSchema schema = new SimpleRecordSchema(fields);
+
+ final Map<String, Object> values = new HashMap<>();
+ values.put("name", "John Doe");
+ values.put("age", 42);
+
+ final String rawForm = "{\"name\":\"John
Doe\",\"age\":42,\"ignoredExtra\":\"preserved\"}";
+ final SerializedForm serializedForm = SerializedForm.of(rawForm,
"application/json");
+ final Record record = new MapRecord(schema, values, serializedForm);
+
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final WriteJsonResult writer = new
WriteJsonResult(Mockito.mock(ComponentLog.class), schema, new
SchemaNameAsAttribute(), baos, false,
+ NullSuppression.NEVER_SUPPRESS, OutputGrouping.OUTPUT_ARRAY,
RecordFieldType.DATE.getDefaultFormat(),
+ RecordFieldType.TIME.getDefaultFormat(),
RecordFieldType.TIMESTAMP.getDefaultFormat(),
+ "application/json", false, false)) {
+ writer.write(RecordSet.of(schema, record));
+ }
+
+ final String output = baos.toString(StandardCharsets.UTF_8);
+ assertFalse(output.contains("ignoredExtra"),
+ "When Serialized JSON Input Handling is DISABLED, the writer
must re-serialize from typed values and ignore raw bytes");
+ assertEquals("[{\"name\":\"John Doe\",\"age\":42}]", output);
+ }
+
+ @Test
+ void testReuseInputSerializationFalseHonorsTimestampFormat() throws
IOException {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new RecordField("event",
RecordFieldType.TIMESTAMP.getDataType()));
+ final RecordSchema schema = new SimpleRecordSchema(fields);
+
+ final Timestamp eventTimestamp = Timestamp.valueOf("2025-03-20
17:33:11.000");
+ final Map<String, Object> values = new HashMap<>();
+ values.put("event", eventTimestamp);
+
+ final String timestampValue = "2025-03-20T17:33:11.000+0000";
+ final String timestampFormat = "yyyy-MM-dd'T'HH:mm:ss.SSSX";
+ final String rawForm = "{\"event\":\"%s\"}".formatted(timestampValue);
+ final SerializedForm serializedForm = SerializedForm.of(rawForm,
"application/json");
+ final Record record = new MapRecord(schema, values, serializedForm);
+
+ final ByteArrayOutputStream fastPathBaos = new ByteArrayOutputStream();
+ try (final WriteJsonResult writer = new
WriteJsonResult(Mockito.mock(ComponentLog.class), schema, new
SchemaNameAsAttribute(), fastPathBaos, false,
+ NullSuppression.NEVER_SUPPRESS, OutputGrouping.OUTPUT_ARRAY,
RecordFieldType.DATE.getDefaultFormat(),
+ RecordFieldType.TIME.getDefaultFormat(), timestampFormat,
+ "application/json", false, true)) {
+ writer.write(RecordSet.of(schema, record));
+ }
+
+
assertTrue(fastPathBaos.toString(StandardCharsets.UTF_8).contains(timestampValue),
+ "With Serialized JSON Input Handling ENABLED, raw '+0000' form
is passed through even though Timestamp Format would normalize to 'Z'");
+
+ final ByteArrayOutputStream slowPathBaos = new ByteArrayOutputStream();
+ try (final WriteJsonResult writer = new
WriteJsonResult(Mockito.mock(ComponentLog.class), schema, new
SchemaNameAsAttribute(), slowPathBaos, false,
+ NullSuppression.NEVER_SUPPRESS, OutputGrouping.OUTPUT_ARRAY,
RecordFieldType.DATE.getDefaultFormat(),
+ RecordFieldType.TIME.getDefaultFormat(), timestampFormat,
+ "application/json", false, false)) {
+ writer.write(RecordSet.of(schema, record));
+ }
+
+ final String slowPathOutput =
slowPathBaos.toString(StandardCharsets.UTF_8);
+ assertFalse(slowPathOutput.contains("+0000"),
+ "With Serialized JSON Input Handling DISABLED, writer's
Timestamp Format must be applied even when SerializedForm is present");
+
assertTrue(slowPathOutput.contains("\"event\":\"2025-03-20T17:33:11.000"),
+ "Re-serialized timestamp should reflect the configured
format");
+ }
+
+ @Test
+ void testReuseInputSerializationFalseHonorsSuppressNulls() throws
IOException {
+ final List<RecordField> fields = new ArrayList<>();
+ fields.add(new RecordField("name",
RecordFieldType.STRING.getDataType()));
+ fields.add(new RecordField("middleName",
RecordFieldType.STRING.getDataType()));
+ final RecordSchema schema = new SimpleRecordSchema(fields);
+
+ final Map<String, Object> values = new HashMap<>();
+ values.put("name", "John Doe");
+ values.put("middleName", null);
+
+ final String rawForm = "{\"name\":\"John Doe\",\"middleName\":null}";
+ final SerializedForm serializedForm = SerializedForm.of(rawForm,
"application/json");
+ final Record record = new MapRecord(schema, values, serializedForm);
+
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (final WriteJsonResult writer = new
WriteJsonResult(Mockito.mock(ComponentLog.class), schema, new
SchemaNameAsAttribute(), baos, false,
+ NullSuppression.ALWAYS_SUPPRESS, OutputGrouping.OUTPUT_ARRAY,
RecordFieldType.DATE.getDefaultFormat(),
+ RecordFieldType.TIME.getDefaultFormat(),
RecordFieldType.TIMESTAMP.getDefaultFormat(),
+ "application/json", false, false)) {
+ writer.write(RecordSet.of(schema, record));
+ }
+
+ final String output = baos.toString(StandardCharsets.UTF_8);
+ assertFalse(output.contains("middleName"),
+ "Suppress Null Values must be honored when Serialized JSON
Input Handling is DISABLED, even though the input JSON contained the null
field");
+ assertEquals("[{\"name\":\"John Doe\"}]", output);
+ }
}
