Event logs for Model

incubator/catalog-apicurio/src/main/java/io/aklivity/zilla/runtime/catalog/apicurio/internal/ApicurioCatalogHandler.java

@@ -177,6 +177,8 @@ public int resolve(
 
     @Override
     public int decode(
+        long traceId,
+        long bindingId,
         DirectBuffer data,
         int index,
         int length,
@@ -195,13 +197,15 @@ public int decode(
 
         if (schemaId > NO_SCHEMA_ID)
         {
-            valLength = decoder.accept(schemaId, data, index + progress, length - progress, next);
+            valLength = decoder.accept(traceId, bindingId, schemaId, data, index + progress, length - progress, next);
         }
         return valLength;
     }
 
     @Override
     public int encode(
+        long traceId,
+        long bindingId,
         int schemaId,
         DirectBuffer data,
         int index,
@@ -211,7 +215,7 @@ public int encode(
     {
         ApicurioPrefixFW prefix = prefixRW.rewrap().schemaId(schemaId).build();
         next.accept(prefix.buffer(), prefix.offset(), prefix.sizeof());
-        int valLength = encoder.accept(schemaId, data, index, length, next);
+        int valLength = encoder.accept(traceId, bindingId, schemaId, data, index, length, next);
         return valLength > 0 ? prefix.sizeof() + valLength : -1;
     }
 

incubator/catalog-apicurio/src/test/java/io/aklivity/zilla/runtime/catalog/apicurio/internal/ApicurioIT.java

@@ -200,7 +200,7 @@ public void shouldVerifyEncodedData()
             0x30, 0x10, 0x70, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
         data.wrap(bytes, 0, bytes.length);
 
-        assertEquals(18, catalog.encode(1, data, 0, data.capacity(),
+        assertEquals(18, catalog.encode(0L, 0L, 1, data, 0, data.capacity(),
             ValueConsumer.NOP, CatalogHandler.Encoder.IDENTITY));
     }
 
@@ -216,7 +216,7 @@ public void shouldResolveSchemaIdAndProcessData()
             0x30, 0x10, 0x70, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
         data.wrap(bytes, 0, bytes.length);
 
-        int valLength = catalog.decode(data, 0, data.capacity(), ValueConsumer.NOP, CatalogHandler.Decoder.IDENTITY);
+        int valLength = catalog.decode(0L, 0L, data, 0, data.capacity(), ValueConsumer.NOP, CatalogHandler.Decoder.IDENTITY);
 
         assertEquals(data.capacity() - 9, valLength);
     }

incubator/catalog-inline/src/test/java/io/aklivity/zilla/runtime/catalog/inline/internal/InlineIT.java

@@ -75,7 +75,7 @@ public void shouldResolveSchemaIdAndProcessData()
         byte[] bytes = payload.getBytes();
         data.wrap(bytes, 0, bytes.length);
 
-        int valLength = catalog.decode(data, 0, data.capacity(), ValueConsumer.NOP, CatalogHandler.Decoder.IDENTITY);
+        int valLength = catalog.decode(0L, 0L, data, 0, data.capacity(), ValueConsumer.NOP, CatalogHandler.Decoder.IDENTITY);
 
         assertEquals(data.capacity(), valLength);
     }
@@ -91,7 +91,7 @@ public void shouldVerifyEncodedData()
             0x30, 0x10, 0x70, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
         data.wrap(bytes, 0, bytes.length);
 
-        assertEquals(13, catalog.encode(1, data, 0, data.capacity(),
+        assertEquals(13, catalog.encode(0L, 0L, 1, data, 0, data.capacity(),
                 ValueConsumer.NOP, CatalogHandler.Encoder.IDENTITY));
     }
 }

incubator/catalog-schema-registry/src/main/java/io/aklivity/zilla/runtime/catalog/schema/registry/internal/SchemaRegistryCatalogHandler.java

@@ -133,6 +133,8 @@ public int resolve(
 
     @Override
     public int decode(
+        long traceId,
+        long bindingId,
         DirectBuffer data,
         int index,
         int length,
@@ -151,13 +153,15 @@ public int decode(
 
         if (schemaId > NO_SCHEMA_ID)
         {
-            valLength = decoder.accept(schemaId, data, index + progress, length - progress, next);
+            valLength = decoder.accept(traceId, bindingId, schemaId, data, index + progress, length - progress, next);
         }
         return valLength;
     }
 
     @Override
     public int encode(
+        long traceId,
+        long bindingId,
         int schemaId,
         DirectBuffer data,
         int index,
@@ -167,7 +171,7 @@ public int encode(
     {
         SchemaRegistryPrefixFW prefix = prefixRW.rewrap().schemaId(schemaId).build();
         next.accept(prefix.buffer(), prefix.offset(), prefix.sizeof());
-        int valLength = encoder.accept(schemaId, data, index, length, next);
+        int valLength = encoder.accept(traceId, bindingId, schemaId, data, index, length, next);
         return valLength > 0 ? prefix.sizeof() + valLength : -1;
     }
 

incubator/catalog-schema-registry/src/test/java/io/aklivity/zilla/runtime/catalog/schema/registry/internal/SchemaRegistryIT.java

@@ -168,7 +168,7 @@ public void shouldVerifyEncodedData()
             0x30, 0x10, 0x70, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
         data.wrap(bytes, 0, bytes.length);
 
-        assertEquals(18, catalog.encode(1, data, 0, data.capacity(),
+        assertEquals(18, catalog.encode(0L, 0L, 1, data, 0, data.capacity(),
             ValueConsumer.NOP, CatalogHandler.Encoder.IDENTITY));
     }
 
@@ -184,7 +184,7 @@ public void shouldResolveSchemaIdAndProcessData()
             0x30, 0x10, 0x70, 0x6f, 0x73, 0x69, 0x74, 0x69, 0x76, 0x65};
         data.wrap(bytes, 0, bytes.length);
 
-        int valLength = catalog.decode(data, 0, data.capacity(), ValueConsumer.NOP, CatalogHandler.Decoder.IDENTITY);
+        int valLength = catalog.decode(0L, 0L, data, 0, data.capacity(), ValueConsumer.NOP, CatalogHandler.Decoder.IDENTITY);
 
         assertEquals(data.capacity() - 5, valLength);
     }

incubator/model-avro.spec/pom.xml

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

incubator/model-avro.spec/src/main/resources/META-INF/zilla/avro_model.idl

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2021-2023 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.
+ */
+scope avro_model
+{
+    scope event
+    {
+        enum AvroModelEventType (uint8)
+        {
+            VALIDATION_FAILED (1)
+        }
+
+        struct AvroModelValidationFailedEx extends core::stream::Extension
+        {
+            string8 error;
+        }
+
+        union AvroModelEventEx switch (AvroModelEventType)
+        {
+            case VALIDATION_FAILED: AvroModelValidationFailedEx validationFailed;
+        }
+    }
+}

incubator/model-avro.spec/src/main/scripts/io/aklivity/zilla/specs/model/avro/config/event.yaml

@@ -0,0 +1,58 @@
+#
+# Copyright 2021-2023 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.
+#
+
+---
+name: test
+telemetry:
+  exporters:
+    exporter0:
+      type: test
+      options:
+        events:
+          - qname: test.net0
+            id: model.avro.validation.failed
+            message: VALIDATION_FAILED java.io.EOFException
+catalogs:
+  test0:
+    type: test
+    options:
+      schema: |
+        {
+          "fields": [
+            {
+              "name": "id",
+              "type": "string"
+            },
+            {
+              "name": "status",
+              "type": "string"
+            }
+          ],
+          "name": "Event",
+          "namespace": "io.aklivity.example",
+          "type": "record"
+        }
+bindings:
+  net0:
+    kind: server
+    type: test
+    options:
+      value:
+        model: avro
+        catalog:
+          catalog0:
+            - subject: test0
+              version: latest
+    exit: app0

incubator/model-avro/pom.xml

@@ -54,11 +54,6 @@
       <groupId>org.apache.avro</groupId>
       <artifactId>avro</artifactId>
     </dependency>
-    <dependency>
-      <groupId>org.jmock</groupId>
-      <artifactId>jmock-junit4</artifactId>
-      <scope>test</scope>
-    </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
@@ -69,6 +64,16 @@
       <artifactId>hamcrest-library</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.kaazing</groupId>
+      <artifactId>k3po.junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.kaazing</groupId>
+      <artifactId>k3po.lang</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>
@@ -82,7 +87,7 @@
         <artifactId>flyweight-maven-plugin</artifactId>
         <version>${project.version}</version>
         <configuration>
-          <scopeNames>core</scopeNames>
+          <scopeNames>core avro_model</scopeNames>
           <packageName>io.aklivity.zilla.runtime.model.avro.internal.types</packageName>
         </configuration>
         <executions>
@@ -208,6 +213,23 @@
           </rules>
         </configuration>
       </plugin>
+      <plugin>
+        <groupId>org.kaazing</groupId>
+        <artifactId>k3po-maven-plugin</artifactId>
+        <dependencies>
+          <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>engine</artifactId>
+            <version>${project.version}</version>
+            <type>test-jar</type>
+          </dependency>
+          <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>engine</artifactId>
+            <version>${project.version}</version>
+          </dependency>
+        </dependencies>
+      </plugin>
     </plugins>
   </build>
 </project>

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

@@ -14,36 +14,33 @@
  */
 package io.aklivity.zilla.runtime.model.avro.internal;
 
-import java.util.function.LongFunction;
-
 import io.aklivity.zilla.runtime.engine.EngineContext;
-import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.ModelConfig;
 import io.aklivity.zilla.runtime.engine.model.ConverterHandler;
 import io.aklivity.zilla.runtime.engine.model.ModelContext;
 import io.aklivity.zilla.runtime.model.avro.config.AvroModelConfig;
 
 public class AvroModelContext implements ModelContext
 {
-    private final LongFunction<CatalogHandler> supplyCatalog;
+    private final EngineContext context;
 
     public AvroModelContext(
         EngineContext context)
     {
-        this.supplyCatalog = context::supplyCatalog;
+        this.context = context;
     }
 
     @Override
     public ConverterHandler supplyReadConverterHandler(
         ModelConfig config)
     {
-        return new AvroReadConverterHandler(AvroModelConfig.class.cast(config), supplyCatalog);
+        return new AvroReadConverterHandler(AvroModelConfig.class.cast(config), context);
     }
 
     @Override
     public ConverterHandler supplyWriteConverterHandler(
         ModelConfig config)
     {
-        return new AvroWriteConverterHandler(AvroModelConfig.class.cast(config), supplyCatalog);
+        return new AvroWriteConverterHandler(AvroModelConfig.class.cast(config), context);
     }
 }