Json serialization of nullable fields (#1341)

Author: akrambek

runtime/model-avro/pom.xml

@@ -191,6 +191,7 @@
         <configuration>
           <excludes>
             <exclude>io/aklivity/zilla/runtime/model/avro/internal/types/**/*.class</exclude>
+            <exclude>org/apache/avro/**/*.class</exclude>
           </excludes>
           <rules>
             <rule>

runtime/model-avro/src/main/java/io/aklivity/zilla/runtime/model/avro/internal/AvroReadConverterHandler.java

@@ -27,6 +27,7 @@
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.io.CanonicalJsonEncoder;
 import org.apache.avro.io.JsonEncoder;
 
 import io.aklivity.zilla.runtime.engine.EngineContext;
@@ -191,7 +192,7 @@ private void deserializeRecord(
                 expandable.wrap(expandable.buffer());
                 record = reader.read(record, decoderFactory.binaryDecoder(in, decoder));
                 Schema schema = record.getSchema();
-                JsonEncoder out = encoderFactory.jsonEncoder(schema, expandable);
+                JsonEncoder out = new CanonicalJsonEncoder(schema, expandable);
                 writer.write(record, out);
                 out.flush();
 

runtime/model-avro/src/main/java/org/apache/avro/io/CanonicalJsonEncoder.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2021-2024 Aklivity Inc
+ *
+ * Licensed under the Aklivity Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ *   https://www.aklivity.io/aklivity-community-license/
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package org.apache.avro.io;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import org.apache.avro.Schema;
+import org.apache.avro.io.parsing.Parser;
+import org.apache.avro.io.parsing.Symbol;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+
+/**
+ * A derived encoder that does the skipping of fields that match the index. It also encodes unions of null and a single
+ * type as a more normal key=value rather than key={type=value}.
+ *
+ * @author zfarkas
+ */
+public final class CanonicalJsonEncoder extends JsonEncoder
+{
+
+
+    public CanonicalJsonEncoder(
+        final Schema sc,
+        final OutputStream out) throws IOException
+    {
+        super(sc, out);
+    }
+
+    public CanonicalJsonEncoder(
+        final Schema sc,
+        final OutputStream out,
+        final boolean pretty) throws IOException
+    {
+        super(sc, out, pretty);
+    }
+
+    public CanonicalJsonEncoder(
+        final Schema sc,
+        final JsonGenerator out) throws IOException
+    {
+        super(sc, out);
+    }
+
+    public Parser getParser()
+    {
+        return parser;
+    }
+
+    public static boolean isNullableSingle(
+        final Symbol.Alternative top)
+    {
+        return top.size() == 2 && ("null".equals(top.getLabel(0)) || "null".equals(top.getLabel(1)));
+    }
+
+    public static String getNullableSingle(
+        final Symbol.Alternative top)
+    {
+        final String label = top.getLabel(0);
+        return "null".equals(label) ? top.getLabel(1) : label;
+    }
+
+    /**
+     * Overwrite this function to optime json decoding of union {null, type}.
+     *
+     * @param unionIndex
+     * @throws IOException
+     */
+
+    @Override
+    public void writeIndex(
+        final int unionIndex) throws IOException
+    {
+        parser.advance(Symbol.UNION);
+        Symbol.Alternative top = (Symbol.Alternative) parser.popSymbol();
+        Symbol symbol = top.getSymbol(unionIndex);
+        if (symbol != Symbol.NULL && !isNullableSingle(top))
+        {
+            out.writeStartObject();
+            out.writeFieldName(top.getLabel(unionIndex));
+            parser.pushSymbol(Symbol.UNION_END);
+        }
+        parser.pushSymbol(symbol);
+    }
+
+}