Addressing review feedback

incubator/catalog-inline/src/main/java/io/aklivity/zilla/runtime/catalog/inline/internal/InlineCatalogHandler.java

@@ -38,12 +38,6 @@ public InlineCatalogHandler(
         registerSchema(config.subjects);
     }
 
-    @Override
-    public String type()
-    {
-        return INLINE;
-    }
-
     @Override
     public int register(
         String subject,

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

@@ -18,21 +18,29 @@
 import java.net.http.HttpClient;
 import java.net.http.HttpRequest;
 import java.net.http.HttpResponse;
+import java.nio.ByteOrder;
 import java.text.MessageFormat;
 import java.util.zip.CRC32C;
 
+import org.agrona.MutableDirectBuffer;
 import org.agrona.collections.Int2ObjectCache;
+import org.agrona.concurrent.UnsafeBuffer;
 
 import io.aklivity.zilla.runtime.catalog.schema.registry.internal.config.SchemaRegistryOptionsConfig;
 import io.aklivity.zilla.runtime.catalog.schema.registry.internal.serializer.RegisterSchemaRequest;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
+import io.aklivity.zilla.runtime.engine.validator.function.ValueConsumer;
 
 public class SchemaRegistryCatalogHandler implements CatalogHandler
 {
     private static final String SUBJECT_VERSION_PATH = "/subjects/{0}/versions/{1}";
     private static final String SCHEMA_PATH = "/schemas/ids/{0}";
     private static final String REGISTER_SCHEMA_PATH = "/subjects/{0}/versions";
+    private static final byte MAGIC_BYTE = 0x0;
+    private static final int ENRICHED_LENGTH = 5;
+    private static final int MAX_PADDING_LEN = 5;
 
+    private final MutableDirectBuffer prefixRO;
     private final HttpClient client;
     private final String baseUrl;
     private final RegisterSchemaRequest request;
@@ -49,12 +57,7 @@ public SchemaRegistryCatalogHandler(
         this.crc32c = new CRC32C();
         this.cache = new Int2ObjectCache<>(1, 1024, i -> {});
         this.schemaIdCache = new Int2ObjectCache<>(1, 1024, i -> {});
-    }
-
-    @Override
-    public String type()
-    {
-        return SCHEMA_REGISTRY;
+        this.prefixRO = new UnsafeBuffer(new byte[5]);
     }
 
     @Override
@@ -130,6 +133,23 @@ public int resolve(
         return schemaId;
     }
 
+    @Override
+    public int enrich(
+        int schemaId,
+        ValueConsumer next)
+    {
+        prefixRO.putByte(0, MAGIC_BYTE);
+        prefixRO.putInt(1, schemaId, ByteOrder.BIG_ENDIAN);
+        next.accept(prefixRO, 0, 5);
+        return ENRICHED_LENGTH;
+    }
+
+    @Override
+    public int maxPadding()
+    {
+        return MAX_PADDING_LEN;
+    }
+
     private String sendHttpRequest(
         String path)
     {

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

@@ -21,6 +21,8 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.rules.RuleChain.outerRule;
 
+import org.agrona.MutableDirectBuffer;
+import org.agrona.concurrent.UnsafeBuffer;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.Test;
@@ -31,6 +33,7 @@
 import org.kaazing.k3po.junit.rules.K3poRule;
 
 import io.aklivity.zilla.runtime.catalog.schema.registry.internal.config.SchemaRegistryOptionsConfig;
+import io.aklivity.zilla.runtime.engine.validator.function.ValueConsumer;
 
 public class SchemaRegistryIT
 {
@@ -153,4 +156,23 @@ public void shouldResolveSchemaViaSubjectVersionFromCache() throws Exception
         assertThat(schema, not(nullValue()));
         assertEquals(expected, schema);
     }
+
+    @Test
+    public void shouldVerifyMaxPadding()
+    {
+        SchemaRegistryCatalogHandler catalog = new SchemaRegistryCatalogHandler(config);
+
+        assertEquals(5, catalog.maxPadding());
+    }
+
+    @Test
+    public void shouldVerifyEnrichedData()
+    {
+        SchemaRegistryCatalogHandler catalog = new SchemaRegistryCatalogHandler(config);
+
+        MutableDirectBuffer value = new UnsafeBuffer(new byte[ 5]);
+        value.putBytes(0, new byte[]{0x00, 0x00, 0x00, 0x00, 0x01});
+
+        assertEquals(value.capacity(), catalog.enrich(1, ValueConsumer.NOP));
+    }
 }

incubator/validator-avro/src/main/java/io/aklivity/zilla/runtime/validator/avro/AvroReadValidator.java

@@ -24,6 +24,9 @@
 import org.agrona.DirectBuffer;
 import org.apache.avro.AvroRuntimeException;
 import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.JsonEncoder;
 
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
@@ -64,7 +67,7 @@ else if (catalog.id != NO_SCHEMA_ID)
             {
                 schemaId = handler.resolve(subject, catalog.version);
             }
-            padding = paddings.computeIfAbsent(schemaId, this::supplyPadding);
+            padding = supplyPadding(schemaId);
         }
         return padding;
     }
@@ -117,7 +120,7 @@ else if (catalog.id != NO_SCHEMA_ID)
             schemaId = handler.resolve(subject, catalog.version);
         }
 
-        reader = readers.computeIfAbsent(schemaId, this::supplyReader);
+        GenericDatumReader<GenericRecord> reader = supplyReader(schemaId);
         if (reader != null)
         {
             int payloadLength = length - progress;
@@ -128,7 +131,7 @@ else if (catalog.id != NO_SCHEMA_ID)
                 int recordLength = record.length;
                 if (recordLength > 0)
                 {
-                    valLength = record.length;
+                    valLength = recordLength;
                     valueRO.wrap(record);
                     next.accept(valueRO, 0, valLength);
                 }
@@ -151,12 +154,13 @@ private byte[] deserializeRecord(
         encoded.reset();
         try
         {
-            record = records.computeIfAbsent(schemaId, this::supplyRecord);
+            GenericDatumReader<GenericRecord> reader = supplyReader(schemaId);
+            GenericDatumWriter<GenericRecord> writer = supplyWriter(schemaId);
+            GenericRecord record = supplyRecord(schemaId);
             in.wrap(buffer, index, length);
             record = reader.read(record, decoderFactory.binaryDecoder(in, decoder));
             Schema schema = record.getSchema();
             JsonEncoder out = encoderFactory.jsonEncoder(schema, encoded);
-            writer = writers.computeIfAbsent(schemaId, this::supplyWriter);
             writer.write(record, out);
             out.flush();
             encoded.close();