support for Avro model and header format change

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/config/KafkaTopicConfigBuilder.java

@@ -20,6 +20,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Function;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.KafkaDeltaType;
 import io.aklivity.zilla.runtime.binding.kafka.internal.types.KafkaOffsetType;
@@ -29,6 +31,11 @@
 
 public final class KafkaTopicConfigBuilder<T> extends ConfigBuilder<T, KafkaTopicConfigBuilder<T>>
 {
+    private static final String PATH = "^\\$\\{message\\.value\\.([A-Za-z_][A-Za-z0-9_]*)\\}$";
+    private static final Pattern PATH_PATTERN = Pattern.compile(PATH);
+    private static final String INTERNAL_PATH_PATTERN = "$.%s";
+
+    private final Matcher matcher;
     private final Function<KafkaTopicConfig, T> mapper;
     private String name;
     private KafkaOffsetType defaultOffset;
@@ -42,6 +49,7 @@ public final class KafkaTopicConfigBuilder<T> extends ConfigBuilder<T, KafkaTopi
     {
         this.mapper = mapper;
         this.headers = new ArrayList<>();
+        this.matcher = PATH_PATTERN.matcher("");
     }
 
     @Override
@@ -94,7 +102,11 @@ public KafkaTopicConfigBuilder<T> header(
         {
             this.headers = new ArrayList<>();
         }
-        this.headers.add(new KafkaTopicHeaderType(new String32FW(name, UTF_8), path));
+        if (matcher.reset(path).matches())
+        {
+            this.headers.add(new KafkaTopicHeaderType(new String32FW(name, UTF_8),
+                String.format(INTERNAL_PATH_PATTERN, matcher.group(1))));
+        }
         return this;
     }
 

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/config/KafkaTopicConfigAdapter.java

@@ -17,11 +17,12 @@
 
 import static jakarta.json.JsonValue.ValueType.OBJECT;
 
+import java.util.Map;
+
 import jakarta.json.Json;
-import jakarta.json.JsonArray;
-import jakarta.json.JsonArrayBuilder;
 import jakarta.json.JsonObject;
 import jakarta.json.JsonObjectBuilder;
+import jakarta.json.JsonString;
 import jakarta.json.JsonValue;
 import jakarta.json.bind.adapter.JsonbAdapter;
 
@@ -79,12 +80,10 @@ public JsonObject adaptToJson(
 
         if (topic.headers != null && !topic.headers.isEmpty())
         {
-            JsonArrayBuilder headers = Json.createArrayBuilder();
+            JsonObjectBuilder headers = Json.createObjectBuilder();
             for (KafkaTopicHeaderType header : topic.headers)
             {
-                JsonObjectBuilder headerJson = Json.createObjectBuilder();
-                headerJson.add(header.name.asString(), header.path);
-                headers.add(headerJson);
+                headers.add(header.name.asString(), header.path);
             }
             object.add(HEADERS_NAME, headers);
         }
@@ -139,17 +138,14 @@ public KafkaTopicConfig adaptFromJson(
             topicBuilder.value(converter.adaptFromJson(valueObject.build()));
         }
 
-        JsonArray headers = object.containsKey(HEADERS_NAME) ? object.getJsonArray(HEADERS_NAME) : null;
+        JsonObject headers = object.containsKey(HEADERS_NAME) ? object.getJsonObject(HEADERS_NAME) : null;
 
-        if (headers != null && !headers.isEmpty())
+        if (headers != null)
         {
-            for (JsonValue header: headers)
+            for (Map.Entry<String, JsonValue> entry : headers.entrySet())
             {
-                JsonObject headerJson = header.asJsonObject();
-                for (String headerName: headerJson.keySet())
-                {
-                    topicBuilder.header(headerName, headerJson.getString(headerName));
-                }
+                JsonString jsonString = (JsonString) entry.getValue();
+                topicBuilder.header(entry.getKey(), jsonString.getString());
             }
         }
 

runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/config/KafkaOptionsConfigAdapterTest.java

@@ -22,6 +22,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.not;
 import static org.hamcrest.Matchers.nullValue;
+import static org.junit.Assert.assertEquals;
 
 import jakarta.json.bind.Jsonb;
 import jakarta.json.bind.JsonbBuilder;
@@ -228,11 +229,9 @@ public void shouldReadHeadersOptions()
                     "{" +
                     "\"name\": \"test\"," +
                     "\"headers\":" +
-                    "[" +
-                        "{" +
-                        "\"correlation-id\": \"$.correlation-id\"" +
-                        "}" +
-                    "]" +
+                    "{" +
+                    "\"correlation-id\": \"${message.value.correlationId}\"" +
+                    "}" +
                     "}" +
                 "]" +
             "}";
@@ -241,10 +240,8 @@ public void shouldReadHeadersOptions()
 
         assertThat(options, not(nullValue()));
         assertThat(options.bootstrap, equalTo(singletonList("test")));
-        assertThat(options.topics, equalTo(singletonList(KafkaTopicConfig.builder()
-            .name("test")
-            .header("correlation-id", "$.correlation-id")
-            .build())));
+        assertEquals(options.topics.get(0).headers.get(0).name.asString(), "correlation-id");
+        assertEquals(options.topics.get(0).headers.get(0).path, "$.correlationId");
     }
 
     @Test
@@ -254,7 +251,7 @@ public void shouldWriteHeadersOptions()
             .bootstrap(singletonList("test"))
             .topics(singletonList(KafkaTopicConfig.builder()
                 .name("test")
-                .header("correlation-id", "$.correlation-id")
+                .header("correlation-id", "${message.value.correlationId}")
                 .value(TestModelConfig.builder().length(0).build())
                 .build()))
             .build();
@@ -263,6 +260,6 @@ public void shouldWriteHeadersOptions()
 
         assertThat(text, not(nullValue()));
         assertThat(text, equalTo("{\"bootstrap\":[\"test\"]," +
-            "\"topics\":[{\"name\":\"test\",\"value\":\"test\",\"headers\":[{\"correlation-id\":\"$.correlation-id\"}]}]}"));
+            "\"topics\":[{\"name\":\"test\",\"value\":\"test\",\"headers\":{\"correlation-id\":\"$.correlationId\"}}]}"));
     }
 }

runtime/model-avro/pom.xml

@@ -85,7 +85,7 @@
         <artifactId>flyweight-maven-plugin</artifactId>
         <version>${project.version}</version>
         <configuration>
-          <scopeNames>core avro_model</scopeNames>
+          <scopeNames>internal core avro_model</scopeNames>
           <packageName>io.aklivity.zilla.runtime.model.avro.internal.types</packageName>
         </configuration>
         <executions>

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

@@ -19,11 +19,15 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.agrona.DirectBuffer;
 import org.agrona.ExpandableDirectByteBuffer;
+import org.agrona.MutableDirectBuffer;
 import org.agrona.collections.Int2IntHashMap;
 import org.agrona.collections.Int2ObjectCache;
+import org.agrona.concurrent.UnsafeBuffer;
 import org.agrona.io.DirectBufferInputStream;
 import org.agrona.io.ExpandableDirectBufferOutputStream;
 import org.apache.avro.AvroRuntimeException;
@@ -42,6 +46,13 @@
 import io.aklivity.zilla.runtime.engine.config.CatalogedConfig;
 import io.aklivity.zilla.runtime.engine.config.SchemaConfig;
 import io.aklivity.zilla.runtime.model.avro.config.AvroModelConfig;
+import io.aklivity.zilla.runtime.model.avro.internal.types.AvroBooleanFW;
+import io.aklivity.zilla.runtime.model.avro.internal.types.AvroBytesFW;
+import io.aklivity.zilla.runtime.model.avro.internal.types.AvroDoubleFW;
+import io.aklivity.zilla.runtime.model.avro.internal.types.AvroFloatFW;
+import io.aklivity.zilla.runtime.model.avro.internal.types.AvroIntFW;
+import io.aklivity.zilla.runtime.model.avro.internal.types.AvroLongFW;
+import io.aklivity.zilla.runtime.model.avro.internal.types.OctetsFW;
 
 public abstract class AvroModelHandler
 {
@@ -54,6 +65,7 @@ public abstract class AvroModelHandler
     private static final int COMMA_LENGTH = ",".length();
     private static final int JSON_FIELD_ARRAY_LENGTH = "\"\":[]," .length() + COMMA_LENGTH * 100;
     private static final int JSON_FIELD_MAP_LENGTH = "\"\":{},".length();
+    private static final DirectBuffer EMPTY_BUFFER = new UnsafeBuffer();
 
     protected final SchemaConfig catalog;
     protected final CatalogHandler handler;
@@ -66,12 +78,21 @@ public abstract class AvroModelHandler
     protected final ExpandableDirectBufferOutputStream expandable;
     protected final DirectBufferInputStream in;
     protected final AvroModelEventContext event;
+    protected final Map<String, OctetsFW> extracted;
 
     private final Int2ObjectCache<Schema> schemas;
     private final Int2ObjectCache<GenericDatumReader<GenericRecord>> readers;
     private final Int2ObjectCache<GenericDatumWriter<GenericRecord>> writers;
     private final Int2ObjectCache<GenericRecord> records;
     private final Int2IntHashMap paddings;
+    private final AvroBytesFW bytesRO;
+    private final AvroIntFW intRO;
+    private final AvroLongFW longRO;
+    private final AvroFloatFW floatRO;
+    private final AvroDoubleFW doubleRO;
+    private final MutableDirectBuffer text;
+
+    private int progress;
 
     protected AvroModelHandler(
         AvroModelConfig config,
@@ -96,6 +117,14 @@ protected AvroModelHandler(
         this.expandable = new ExpandableDirectBufferOutputStream(new ExpandableDirectByteBuffer());
         this.in = new DirectBufferInputStream();
         this.event = new AvroModelEventContext(context);
+        this.extracted = new HashMap<>();
+        this.bytesRO = new AvroBytesFW();
+        this.intRO = new AvroIntFW();
+        this.longRO = new AvroLongFW();
+        this.floatRO = new AvroFloatFW();
+        this.doubleRO = new AvroDoubleFW();
+        this.text = new UnsafeBuffer(new byte[24]);
+
     }
 
     protected final boolean validate(
@@ -107,16 +136,25 @@ protected final boolean validate(
         int length)
     {
         boolean status = false;
+        for (OctetsFW value: extracted.values())
+        {
+            value.wrap(EMPTY_BUFFER, 0, 0);
+        }
         try
         {
             GenericRecord record = supplyRecord(schemaId);
             in.wrap(buffer, index, length);
             GenericDatumReader<GenericRecord> reader = supplyReader(schemaId);
+            Schema schema = supplySchema(schemaId);
             if (reader != null)
             {
-                reader.read(record, decoderFactory.binaryDecoder(in, decoder));
+                decoderFactory.binaryDecoder(in, decoder);
+                reader.read(record, decoder);
                 status = true;
+
             }
+            progress = index;
+            extractFields(buffer, length, schema);
         }
         catch (IOException | AvroRuntimeException ex)
         {
@@ -125,6 +163,17 @@ protected final boolean validate(
         return status;
     }
 
+    private void extractFields(
+        DirectBuffer buffer,
+        int length,
+        Schema schema)
+    {
+        for (Schema.Field field : schema.getFields())
+        {
+            extract(field.schema(), buffer, length, extracted.get(field.name()));
+        }
+    }
+
     protected final Schema supplySchema(
         int schemaId)
     {
@@ -250,4 +299,97 @@ private int calculatePadding(
         }
         return padding;
     }
+
+    private void extract(
+        Schema schema,
+        DirectBuffer data,
+        int limit,
+        OctetsFW octets)
+    {
+        switch (schema.getType())
+        {
+        case RECORD:
+            extractFields(data, limit, schema);
+            break;
+        case BYTES:
+        case STRING:
+            AvroBytesFW bytes = bytesRO.wrap(data, progress, limit);
+            OctetsFW value = bytes.value();
+            progress = bytes.limit();
+            if (octets != null)
+            {
+                octets.wrap(value.buffer(), value.offset(), value.limit());
+            }
+            break;
+        case INT:
+            AvroIntFW int32 = intRO.wrap(data, progress, limit);
+            int intValue = int32.value();
+            progress = int32.limit();
+            int length = text.putIntAscii(0, intValue);
+            if (octets != null)
+            {
+                octets.wrap(text, 0, length);
+            }
+            break;
+        case FLOAT:
+            AvroFloatFW avroFloat = floatRO.wrap(data, progress, limit);
+            int len = 0;
+            DirectBuffer buffer = avroFloat.value().value();
+            float floatValue = Float.intBitsToFloat(decodeNumberBytes(len, buffer));
+            progress = avroFloat.limit();
+            length = text.putStringWithoutLengthAscii(0, String.valueOf(floatValue));
+            if (octets != null)
+            {
+                octets.wrap(text, 0, length);
+            }
+            break;
+        case LONG:
+            AvroLongFW avroLong = longRO.wrap(data, progress, limit);
+            long longValue = avroLong.value();
+            progress = avroLong.limit();
+            length = text.putLongAscii(0, longValue);
+            if (octets != null)
+            {
+                octets.wrap(text, 0, length);
+            }
+            break;
+        case DOUBLE:
+            AvroDoubleFW avroDouble = doubleRO.wrap(data, progress, limit);
+            len = 0;
+            buffer = avroDouble.value().value();
+            int decoded = (buffer.getByte(len++) & 0xff) |
+                ((buffer.getByte(len++) & 0xff) << 8) |
+                ((buffer.getByte(len++) & 0xff) << 16) |
+                ((buffer.getByte(len++) & 0xff) << 24);
+
+            double doubleValue = Double.longBitsToDouble((((long) decoded) & 0xffffffffL) |
+                (((long) decodeNumberBytes(len, buffer)) << 32));
+            progress = avroDouble.limit();
+            length = text.putStringWithoutLengthAscii(0, String.valueOf(doubleValue));
+            if (octets != null)
+            {
+                octets.wrap(text, 0, length);
+            }
+            break;
+        case BOOLEAN:
+            AvroBooleanFW avroBoolean = new AvroBooleanFW().wrap(data, progress, limit);
+            value = avroBoolean.value();
+            progress = avroBoolean.limit();
+            if (octets != null)
+            {
+                octets.wrap(value.buffer(), value.offset(), value.limit());
+            }
+            break;
+        }
+    }
+
+    private static int decodeNumberBytes(
+        int len,
+        DirectBuffer buffer)
+    {
+        return (buffer.getByte(len++) & 0xff) |
+            ((buffer.getByte(len++) & 0xff) << 8) |
+            ((buffer.getByte(len++) & 0xff) << 16) |
+            ((buffer.getByte(len) & 0xff) << 24);
+    }
 }