Catalog Handler interface to support dynamic encode padding by length

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

@@ -375,7 +375,8 @@ public int encode(
     }
 
     @Override
-    public int encodePadding()
+    public int encodePadding(
+        int length)
     {
         return MAX_PADDING_LENGTH;
     }

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

@@ -164,7 +164,7 @@ public void shouldVerifyMaxPadding()
     {
         ApicurioCatalogHandler catalog = new ApicurioCatalogHandler(config, context, 0L);
 
-        assertEquals(9, catalog.encodePadding());
+        assertEquals(9, catalog.encodePadding(0));
     }
 
     @Test

runtime/catalog-karapace/src/main/java/io/aklivity/zilla/runtime/catalog/karapace/internal/KarapaceCatalogHandler.java

@@ -331,7 +331,8 @@ public int encode(
     }
 
     @Override
-    public int encodePadding()
+    public int encodePadding(
+        int length)
     {
         return MAX_PADDING_LENGTH;
     }

runtime/catalog-karapace/src/test/java/io/aklivity/zilla/runtime/catalog/karapace/internal/KarapaceIT.java

@@ -164,7 +164,7 @@ public void shouldVerifyMaxPadding()
     {
         KarapaceCatalogHandler catalog = new KarapaceCatalogHandler(config, context, 0L);
 
-        assertEquals(5, catalog.encodePadding());
+        assertEquals(5, catalog.encodePadding(0));
     }
 
     @Test

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

@@ -101,7 +101,16 @@ default int encode(
         return encoder.accept(traceId, bindingId, schemaId, data, index, length, next);
     }
 
-    default int encodePadding()
+    default int encodePadding(
+        int length)
+    {
+        return 0;
+    }
+
+    default int decodePadding(
+        DirectBuffer data,
+        int index,
+        int length)
     {
         return 0;
     }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/model/TestConverterHandler.java

@@ -54,7 +54,7 @@ public int padding(
         int index,
         int length)
     {
-        return handler.encodePadding();
+        return handler.encodePadding(length);
     }
 
     @Override

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

@@ -46,7 +46,7 @@ public int padding(
         int index,
         int length)
     {
-        int padding = 0;
+        int padding = handler.decodePadding(data, index, length);
         if (VIEW_JSON.equals(view))
         {
             int schemaId = handler.resolve(data, index, length);

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

@@ -44,7 +44,7 @@ public int padding(
         int index,
         int length)
     {
-        return handler.encodePadding();
+        return handler.encodePadding(length);
     }
 
     @Override

runtime/model-json/src/main/java/io/aklivity/zilla/runtime/model/json/internal/JsonReadConverterHandler.java

@@ -32,6 +32,15 @@ public JsonReadConverterHandler(
         super(config, context);
     }
 
+    @Override
+    public int padding(
+        DirectBuffer data,
+        int index,
+        int length)
+    {
+        return handler.decodePadding(data, index, length);
+    }
+
     @Override
     public int convert(
         long traceId,

runtime/model-json/src/main/java/io/aklivity/zilla/runtime/model/json/internal/JsonWriteConverterHandler.java

@@ -37,7 +37,7 @@ public int padding(
         int index,
         int length)
     {
-        return handler.encodePadding();
+        return handler.encodePadding(length);
     }
 
     @Override

runtime/model-protobuf/src/main/java/io/aklivity/zilla/runtime/model/protobuf/internal/ProtobufReadConverterHandler.java

@@ -53,7 +53,7 @@ public int padding(
         int index,
         int length)
     {
-        int padding = 0;
+        int padding = handler.decodePadding(data, index, length);
         if (VIEW_JSON.equals(view))
         {
             int schemaId = handler.resolve(data, index, length);

runtime/model-protobuf/src/main/java/io/aklivity/zilla/runtime/model/protobuf/internal/ProtobufWriteConverterHandler.java

@@ -58,7 +58,7 @@ public int padding(
                 ? catalog.id
                 : handler.resolve(subject, catalog.version);
 
-        return handler.encodePadding() + supplyIndexPadding(schemaId);
+        return handler.encodePadding(length) + supplyIndexPadding(schemaId);
     }
 
     @Override