NIFI-15856 - JsonRecordSetWriter silently ignores Timestamp Format

pvillard31 · pvillard31 · commit e886a1a8c623 · 2026-04-17T14:36:49.000+02:00
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
@@ -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 reuseInputSerialization;
 
     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 reuseInputSerialization) throws IOException {
 
         super(out);
         this.logger = logger;
@@ -87,6 +94,7 @@ public WriteJsonResult(final ComponentLog logger, final RecordSchema recordSchem
         this.outputGrouping = outputGrouping;
         this.mimeType = mimeType;
         this.allowScientificNotation = allowScientificNotation;
+        this.reuseInputSerialization = reuseInputSerialization;
 
         this.dateFormat = dateFormat;
         this.timeFormat = timeFormat;
@@ -171,6 +179,10 @@ public WriteResult writeRawRecord(final Record record) throws IOException {
     }
 
     private boolean isUseSerializeForm(final Record record, final RecordSchema writeSchema) {
+        if (!reuseInputSerialization) {
+            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
@@ -123,13 +123,28 @@ 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 REUSE_INPUT_SERIALIZATION = new PropertyDescriptor.Builder()
+            .name("Reuse Input Serialization")
+            .description("""
+                    Controls a throughput optimization that only applies to pure JSON pass-through flows. When set to true, and all of the \
+                    following conditions are met, the writer will emit the record's original JSON bytes verbatim instead of re-serializing from typed field \
+                    values: (1) the upstream reader is JsonTreeReader, (2) no data change, (3) the reader's and writer's record schemas are identical, \
+                    (4) the writer is not using compression, and (5) the Pretty Print JSON and Allow Scientific Notation settings are compatible with the \
+                    cached bytes. When the optimization kicks in, the Timestamp Format, Date Format, Time Format, and Suppress Null Values properties \
+                    have no effect on those records. Set this to false to have those properties honored uniformly for every record.""")
+            .expressionLanguageSupported(ExpressionLanguageScope.NONE)
+            .allowableValues("true", "false")
+            .defaultValue("true")
+            .required(true)
+            .build();
 
     private volatile boolean prettyPrint;
     private volatile boolean allowScientificNotation;
     private volatile NullSuppression nullSuppression;
     private volatile OutputGrouping outputGrouping;
     private volatile String compressionFormat;
     private volatile int compressionLevel;
+    private volatile boolean reuseInputSerialization;
 
     @Override
     protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
@@ -140,6 +155,7 @@ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
         properties.add(OUTPUT_GROUPING);
         properties.add(COMPRESSION_FORMAT);
         properties.add(COMPRESSION_LEVEL);
+        properties.add(REUSE_INPUT_SERIALIZATION);
         return properties;
     }
 
@@ -197,6 +213,7 @@ public void onEnabled(final ConfigurationContext context) {
 
         this.compressionFormat = context.getProperty(COMPRESSION_FORMAT).getValue();
         this.compressionLevel = context.getProperty(COMPRESSION_LEVEL).asInteger();
+        this.reuseInputSerialization = context.getProperty(REUSE_INPUT_SERIALIZATION).asBoolean();
     }
 
     @Override
@@ -241,7 +258,7 @@ public RecordSetWriter createWriter(final ComponentLog logger, final RecordSchem
         }
 
         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, reuseInputSerialization);
     }
 
 }
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
@@ -52,6 +52,7 @@
 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,129 @@ void testChoiceArrayOfStringsOrArrayOfRecords() throws IOException {
         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);
+        assertTrue(output.contains("\"ignoredExtra\":\"preserved\""),
+                "Default constructor preserves legacy fast-path behavior: raw bytes should be emitted verbatim");
+    }
+
+    @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 Reuse Input Serialization is false, 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 rawForm = "{\"event\":\"2025-03-20T17:33:11.000+0000\"}";
+        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(), "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+                "application/json", false, true)) {
+            writer.write(RecordSet.of(schema, record));
+        }
+
+        assertTrue(fastPathBaos.toString(StandardCharsets.UTF_8).contains("+0000"),
+                "With Reuse Input Serialization 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(), "yyyy-MM-dd'T'HH:mm:ss.SSSX",
+                "application/json", false, false)) {
+            writer.write(RecordSet.of(schema, record));
+        }
+
+        final String slowPathOutput = slowPathBaos.toString(StandardCharsets.UTF_8);
+        assertFalse(slowPathOutput.contains("+0000"),
+                "With Reuse Input Serialization 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 Reuse Input Serialization is false, even though the input JSON contained the null field");
+        assertEquals("[{\"name\":\"John Doe\"}]", output);
+    }
 }
