addressing review comments

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

@@ -38,7 +38,7 @@ public class SchemaRegistryCatalogHandler implements CatalogHandler
     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 static final int MAX_PADDING_LENGTH = 5;
 
     private final MutableDirectBuffer prefixRO;
     private final HttpClient client;
@@ -147,7 +147,7 @@ public int enrich(
     @Override
     public int maxPadding()
     {
-        return MAX_PADDING_LEN;
+        return MAX_PADDING_LENGTH;
     }
 
     private String sendHttpRequest(

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

@@ -127,12 +127,13 @@ else if (catalog.id != NO_SCHEMA_ID)
             int payloadIndex = index + progress;
             if (FORMAT_JSON.equals(format))
             {
-                byte[] record = deserializeRecord(schemaId, data, payloadIndex, payloadLength);
-                int recordLength = record.length;
+                int recordLength = encoded.position();
+                deserializeRecord(schemaId, data, payloadIndex, payloadLength);
+                recordLength = encoded.position() - recordLength;
                 if (recordLength > 0)
                 {
                     valLength = recordLength;
-                    valueRO.wrap(record);
+                    valueRO.wrap(encoded.buffer());
                     next.accept(valueRO, 0, valLength);
                 }
             }
@@ -145,13 +146,12 @@ else if (validate(schemaId, data, payloadIndex, payloadLength))
         return valLength;
     }
 
-    private byte[] deserializeRecord(
+    private void deserializeRecord(
         int schemaId,
         DirectBuffer buffer,
         int index,
         int length)
     {
-        encoded.reset();
         try
         {
             GenericDatumReader<GenericRecord> reader = supplyReader(schemaId);
@@ -169,6 +169,5 @@ record = reader.read(record, decoderFactory.binaryDecoder(in, decoder));
         {
             ex.printStackTrace();
         }
-        return encoded.toByteArray();
     }
 }

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

@@ -18,20 +18,24 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.OutputStream;
 import java.util.function.LongFunction;
 
 import org.agrona.DirectBuffer;
+import org.agrona.ExpandableDirectByteBuffer;
 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;
 import org.apache.avro.Schema;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.io.BinaryEncoder;
 import org.apache.avro.io.DecoderFactory;
 import org.apache.avro.io.EncoderFactory;
 
@@ -45,7 +49,8 @@ public abstract class AvroValidator
     protected static final byte MAGIC_BYTE = 0x0;
     protected static final String FORMAT_JSON = "json";
 
-    private static final InputStream EMPTY_STREAM = new ByteArrayInputStream(new byte[0]);
+    private static final InputStream EMPTY_INPUT_STREAM = new ByteArrayInputStream(new byte[0]);
+    private static final OutputStream EMPTY_OUTPUT_STREAM = new ByteArrayOutputStream(0);
     private static final int JSON_FIELD_STRUCTURE_LENGTH = "\"\":\"\",".length();
 
     protected final DirectBuffer valueRO;
@@ -54,9 +59,10 @@ public abstract class AvroValidator
     protected final DecoderFactory decoderFactory;
     protected final EncoderFactory encoderFactory;
     protected final BinaryDecoder decoder;
+    protected final BinaryEncoder encoder;
     protected final String subject;
     protected final String format;
-    protected final ByteArrayOutputStream encoded;
+    protected final ExpandableDirectBufferOutputStream encoded;
     protected final DirectBufferInputStream in;
 
     private final Int2ObjectCache<Schema> schemas;
@@ -70,8 +76,9 @@ protected AvroValidator(
         LongFunction<CatalogHandler> supplyCatalog)
     {
         this.decoderFactory = DecoderFactory.get();
-        this.decoder = decoderFactory.binaryDecoder(EMPTY_STREAM, null);
+        this.decoder = decoderFactory.binaryDecoder(EMPTY_INPUT_STREAM, null);
         this.encoderFactory = EncoderFactory.get();
+        this.encoder = encoderFactory.binaryEncoder(EMPTY_OUTPUT_STREAM, null);
         CatalogedConfig cataloged = config.cataloged.get(0);
         this.handler = supplyCatalog.apply(cataloged.id);
         this.catalog = cataloged.schemas.size() != 0 ? cataloged.schemas.get(0) : null;
@@ -85,7 +92,7 @@ protected AvroValidator(
         this.records = new Int2ObjectCache<>(1, 1024, i -> {});
         this.paddings = new Int2IntHashMap(-1);
         this.valueRO = new UnsafeBuffer();
-        this.encoded = new ByteArrayOutputStream();
+        this.encoded = new ExpandableDirectBufferOutputStream(new ExpandableDirectByteBuffer());
         this.in = new DirectBufferInputStream();
     }
 
@@ -100,7 +107,7 @@ protected final boolean validate(
         {
             GenericRecord record = supplyRecord(schemaId);
             in.wrap(buffer, index, length);
-            GenericDatumReader<GenericRecord> reader = readers.computeIfAbsent(schemaId, this::supplyReader);
+            GenericDatumReader<GenericRecord> reader = supplyReader(schemaId);
             reader.read(record, decoderFactory.binaryDecoder(in, decoder));
             status = true;
         }
@@ -126,24 +133,25 @@ protected final int supplyPadding(
     protected final GenericDatumReader<GenericRecord> supplyReader(
         int schemaId)
     {
-        return readers.computeIfAbsent(schemaId, id -> createReader(supplySchema(id)));
+        return readers.computeIfAbsent(schemaId, this::createReader);
     }
 
     protected final GenericDatumWriter<GenericRecord> supplyWriter(
         int schemaId)
     {
-        return writers.computeIfAbsent(schemaId, id -> createWriter(supplySchema(id)));
+        return writers.computeIfAbsent(schemaId, this::createWriter);
     }
 
     protected final GenericRecord supplyRecord(
         int schemaId)
     {
-        return records.computeIfAbsent(schemaId, id -> createRecord(supplySchema(schemaId)));
+        return records.computeIfAbsent(schemaId, this::createRecord);
     }
 
     private GenericDatumReader<GenericRecord> createReader(
-        Schema schema)
+        int schemaId)
     {
+        Schema schema = supplySchema(schemaId);
         GenericDatumReader<GenericRecord> reader = null;
         if (schema != null)
         {
@@ -153,8 +161,9 @@ private GenericDatumReader<GenericRecord> createReader(
     }
 
     private GenericDatumWriter<GenericRecord> createWriter(
-        Schema schema)
+        int schemaId)
     {
+        Schema schema = supplySchema(schemaId);
         GenericDatumWriter<GenericRecord> writer = null;
         if (schema != null)
         {
@@ -164,8 +173,9 @@ private GenericDatumWriter<GenericRecord> createWriter(
     }
 
     private GenericRecord createRecord(
-        Schema schema)
+        int schemaId)
     {
+        Schema schema = supplySchema(schemaId);
         GenericRecord record = null;
         if (schema != null)
         {

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

@@ -23,7 +23,6 @@
 import org.apache.avro.generic.GenericDatumReader;
 import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.io.BinaryEncoder;
 
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.validator.FragmentValidator;
@@ -91,13 +90,13 @@ private int validateComplete(
         {
             if (FORMAT_JSON.equals(format))
             {
-                byte[] record = serializeJsonRecord(schemaId, data, index, length);
-
-                int recordLength = record.length;
+                int recordLength = encoded.position();
+                serializeJsonRecord(schemaId, data, index, length);
+                recordLength = encoded.position() - recordLength;
                 if (recordLength > 0)
                 {
                     valLength = recordLength + handler.enrich(schemaId, next);
-                    valueRO.wrap(record);
+                    valueRO.wrap(encoded.buffer());
                     next.accept(valueRO, 0, recordLength);
                 }
             }
@@ -110,13 +109,12 @@ else if (validate(schemaId, data, index, length))
         return valLength;
     }
 
-    private byte[] serializeJsonRecord(
+    private void serializeJsonRecord(
         int schemaId,
         DirectBuffer buffer,
         int index,
         int length)
     {
-        encoded.reset();
         try
         {
             Schema schema = supplySchema(schemaId);
@@ -125,15 +123,14 @@ private byte[] serializeJsonRecord(
             GenericRecord record = supplyRecord(schemaId);
             in.wrap(buffer, index, length);
             record = reader.read(record, decoderFactory.jsonDecoder(schema, in));
-            BinaryEncoder out = encoderFactory.binaryEncoder(encoded, null);
-            writer.write(record, out);
-            out.flush();
+            encoderFactory.binaryEncoder(encoded, encoder);
+            writer.write(record, encoder);
+            encoder.flush();
             encoded.close();
         }
         catch (IOException | AvroRuntimeException ex)
         {
             ex.printStackTrace();
         }
-        return encoded.toByteArray();
     }
 }

incubator/validator-avro/src/test/java/io/aklivity/zilla/runtime/validator/avro/AvroValidatorTest.java

@@ -73,7 +73,6 @@ public void shouldVerifyValidAvroEvent()
         CatalogConfig catalogConfig = new CatalogConfig("test0", "test",
             TestCatalogOptionsConfig.builder()
                 .id(9)
-                .exclude(true)
                 .schema(SCHEMA)
                 .build());
         LongFunction<CatalogHandler> handler = value -> context.attach(catalogConfig);
@@ -173,7 +172,6 @@ public void shouldReadAvroEventExpectJson()
         CatalogConfig catalogConfig = new CatalogConfig("test0", "test",
             TestCatalogOptionsConfig.builder()
                 .id(9)
-                .exclude(true)
                 .schema(SCHEMA)
                 .build());
         LongFunction<CatalogHandler> handler = value -> context.attach(catalogConfig);
@@ -207,6 +205,8 @@ public void shouldReadAvroEventExpectJson()
 
         int progress = validator.validate(data, 0, data.capacity(), ValueConsumer.NOP);
         assertEquals(expected.capacity(), progress);
+
+        assertEquals(expected.capacity(), validator.validate(data, 0, data.capacity(), ValueConsumer.NOP));
     }
 
     @Test
@@ -248,6 +248,8 @@ public void shouldWriteJsonEventExpectAvro()
         data.wrap(payload.getBytes(), 0, payload.getBytes().length);
         int progress = validator.validate(data, 0, data.capacity(), ValueConsumer.NOP);
         assertEquals(expected.capacity(), progress);
+
+        assertEquals(expected.capacity(), validator.validate(data, 0, data.capacity(), ValueConsumer.NOP));
     }
 
     @Test

incubator/validator-core/src/main/java/io/aklivity/zilla/runtime/validator/core/StringEncoding.java

@@ -27,18 +27,20 @@ public boolean validate(
             int length)
         {
             final int limit = index + length;
+            validate:
             while (index < limit)
             {
                 final int charByte0 = data.getByte(index);
                 final int charByteCount = (charByte0 & 0b1000_0000) != 0
                     ? Integer.numberOfLeadingZeros((~charByte0 & 0xff) << 24)
                     : 1;
 
-                for (int j = 1; j < charByteCount; j++)
+                final int charByteLimit = index + charByteCount;
+                for (int charByteIndex = index + 1; charByteIndex < charByteLimit; charByteIndex++)
                 {
-                    if (index + j >= limit || (data.getByte(index + j) & 0b11000000) != 0b10000000)
+                    if (charByteIndex >= limit || (data.getByte(charByteIndex) & 0b11000000) != 0b10000000)
                     {
-                        break;
+                        break validate;
                     }
                 }
                 index += charByteCount;

incubator/validator-json/src/main/java/io/aklivity/zilla/runtime/validator/json/JsonValidator.java

@@ -98,7 +98,7 @@ private JsonSchema supplySchema(
     private JsonProvider supplyProvider(
         int schemaId)
     {
-        return providers.computeIfAbsent(schemaId, id -> createProvider(supplySchema(id)));
+        return providers.computeIfAbsent(schemaId, this::createProvider);
     }
 
     private JsonSchema resolveSchema(
@@ -117,8 +117,9 @@ private JsonSchema resolveSchema(
     }
 
     private JsonProvider createProvider(
-        JsonSchema schema)
+        int schemaId)
     {
+        JsonSchema schema = supplySchema(schemaId);
         JsonProvider provider = null;
         if (schema != null)
         {

runtime/engine/src/main/java/io/aklivity/zilla/runtime/engine/catalog/CatalogHandler.java

@@ -20,8 +20,6 @@
 public interface CatalogHandler
 {
     int NO_SCHEMA_ID = 0;
-    int NO_ENRICHMENT = 0;
-    int ZERO_PADDING = 0;
 
     int register(
         String subject,
@@ -39,11 +37,11 @@ default int enrich(
         int schemaId,
         ValueConsumer next)
     {
-        return NO_ENRICHMENT;
+        return 0;
     }
 
     default int maxPadding()
     {
-        return ZERO_PADDING;
+        return 0;
     }
 }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/catalog/TestCatalogHandler.java

@@ -26,15 +26,14 @@
 
 public class TestCatalogHandler implements CatalogHandler
 {
-    private static final int MAX_PADDING_LEN = 10;
+    private static final int MAX_PADDING_LENGTH = 10;
     private static final byte MAGIC_BYTE = 0x0;
-    private static final int ENRICHED_LENGTH = 5;
+    private static final int PREFIX_LENGTH = 5;
 
     private final String schema;
     private final MutableDirectBuffer prefixRO;
     private final int id;
     private final boolean embed;
-    private final boolean exclude;
 
     public TestCatalogHandler(
         TestCatalogOptionsConfig config)
@@ -43,7 +42,6 @@ public TestCatalogHandler(
         this.prefixRO = new UnsafeBuffer(new byte[5]);
         this.id = config.id;
         this.embed = config.embed;
-        this.exclude = config.exclude;
     }
 
     @Override
@@ -56,20 +54,16 @@ public int enrich(
         {
             prefixRO.putByte(0, MAGIC_BYTE);
             prefixRO.putInt(1, schemaId, ByteOrder.BIG_ENDIAN);
-            next.accept(prefixRO, 0, 5);
-            length = ENRICHED_LENGTH;
-        }
-        else if (exclude)
-        {
-            length = ENRICHED_LENGTH;
+            next.accept(prefixRO, 0, PREFIX_LENGTH);
+            length = PREFIX_LENGTH;
         }
         return length;
     }
 
     @Override
     public int maxPadding()
     {
-        return MAX_PADDING_LEN;
+        return MAX_PADDING_LENGTH;
     }
 
     @Override