Support catalog register new schema version

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

@@ -42,6 +42,7 @@
 public class SchemaRegistryCatalogHandler implements CatalogHandler
 {
     private static final String SUBJECT_VERSION_PATH = "/subjects/{0}/versions/{1}";
+    private static final String SUBJECT_PATH = "/subjects/{0}/versions";
     private static final String SCHEMA_PATH = "/schemas/ids/{0}";
 
     private static final int MAX_PADDING_LENGTH = 5;
@@ -80,6 +81,22 @@ public SchemaRegistryCatalogHandler(
         this.cachedSchemaIds = catalog.cache.schemaIds;
     }
 
+    @Override
+    public int register(
+        String subject,
+        String schema)
+    {
+        int versionId = NO_VERSION_ID;
+
+        String response = sendPostHttpRequest(MessageFormat.format(SUBJECT_PATH, subject), schema);
+        if (response != null)
+        {
+            versionId = request.resolveResponse(response);
+        }
+
+        return versionId;
+    }
+
     @Override
     public String resolve(
         int schemaId)
@@ -348,6 +365,30 @@ private String sendHttpRequest(
         return responseBody;
     }
 
+    private String sendPostHttpRequest(
+        String path,
+        String body)
+    {
+        HttpRequest httpRequest = HttpRequest
+            .newBuilder(toURI(baseUrl, path))
+            .header("content-type", "application/json")
+            .POST(HttpRequest.BodyPublishers.ofString(body))
+            .build();
+        // TODO: introduce interrupt/timeout for request to schema registry
+
+        String responseBody;
+        try
+        {
+            HttpResponse<String> httpResponse = client.send(httpRequest, HttpResponse.BodyHandlers.ofString());
+            responseBody = httpResponse.statusCode() == 200 ? httpResponse.body() : null;
+        }
+        catch (Exception ex)
+        {
+            responseBody = null;
+        }
+        return responseBody;
+    }
+
     @Override
     public String location()
     {

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

@@ -134,4 +134,15 @@ public void shouldResolveSchemaIdFromCacheAndRetry() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("register/zilla.yaml")
+    @Specification({
+        "${net}/handshake/client",
+        "${app}/handshake/server",
+        "${remote}/register.schema" })
+    public void shouldRegisterSchema() throws Exception
+    {
+        k3po.finish();
+    }
 }

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

@@ -22,6 +22,7 @@
 public interface CatalogHandler
 {
     int NO_SCHEMA_ID = 0;
+    int NO_VERSION_ID = 0;
 
     @FunctionalInterface
     interface Decoder
@@ -61,6 +62,13 @@ int accept(
             ValueConsumer next);
     }
 
+    default int register(
+        String subject,
+        String schema)
+    {
+        return NO_VERSION_ID;
+    }
+
     String resolve(
         int schemaId);
 

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/binding/TestBindingFactory.java

@@ -32,7 +32,6 @@
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
 import io.aklivity.zilla.runtime.engine.config.CatalogedConfig;
-import io.aklivity.zilla.runtime.engine.config.RouteConfig;
 import io.aklivity.zilla.runtime.engine.config.SchemaConfig;
 import io.aklivity.zilla.runtime.engine.guard.GuardHandler;
 import io.aklivity.zilla.runtime.engine.model.ConverterHandler;
@@ -85,8 +84,9 @@ final class TestBindingFactory implements BindingHandler
     private final TestEventContext event;
 
     private ConverterHandler valueType;
-    private List<CatalogHandler> catalogs;
+    private String schema;
     private SchemaConfig catalog;
+    private List<CatalogHandler> catalogs;
     private List<CatalogAssertion> catalogAssertions;
     private GuardHandler guard;
     private String credentials;
@@ -106,15 +106,9 @@ final class TestBindingFactory implements BindingHandler
     public void attach(
         BindingConfig binding)
     {
-        RouteConfig exit = binding.routes.stream()
+        binding.routes.stream()
             .filter(r -> r.when.isEmpty())
-            .findFirst()
-            .orElse(null);
-
-        if (exit != null)
-        {
-            router.put(binding.id, exit.id);
-        }
+            .findFirst().ifPresent(exit -> router.put(binding.id, exit.id));
 
         TestBindingOptionsConfig options = (TestBindingOptionsConfig) binding.options;
         if (options != null)
@@ -124,9 +118,11 @@ public void attach(
                 this.valueType = context.supplyWriteConverter(options.value);
             }
 
+            this.schema = options.schema;
+
             if (options.cataloged != null)
             {
-                this.catalog = options.cataloged.size() != 0 ? options.cataloged.get(0).schemas.get(0) : null;
+                this.catalog = !options.cataloged.isEmpty() ? options.cataloged.get(0).schemas.get(0) : null;
                 this.catalogs = new LinkedList<>();
                 for (CatalogedConfig catalog : options.cataloged)
                 {
@@ -347,7 +343,11 @@ else if (assertion.schema != null && !assertion.schema.equals(schema))
                 }
                 else
                 {
-                    if (catalog.subject != null && catalog.version != null)
+                    if (catalog.subject != null && schema != null)
+                    {
+                        handler.register(catalog.subject, schema);
+                    }
+                    else if (catalog.subject != null && catalog.version != null)
                     {
                         handler.resolve(catalog.subject, catalog.version);
                     }

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/binding/config/TestBindingOptionsConfig.java

@@ -26,6 +26,7 @@ public final class TestBindingOptionsConfig extends OptionsConfig
 {
     public final ModelConfig value;
     public final String mode;
+    public final String schema;
     public final TestAuthorizationConfig authorization;
     public final List<CatalogedConfig> cataloged;
     public final List<Event> events;
@@ -46,6 +47,7 @@ public static <T> TestBindingOptionsConfigBuilder<T> builder(
     TestBindingOptionsConfig(
         ModelConfig value,
         String mode,
+        String schema,
         TestAuthorizationConfig authorization,
         List<CatalogedConfig> cataloged,
         List<Event> events,
@@ -55,6 +57,7 @@ public static <T> TestBindingOptionsConfigBuilder<T> builder(
         super(value != null ? List.of(value) : List.of(), List.of());
         this.value = value;
         this.mode = mode;
+        this.schema = schema;
         this.authorization = authorization;
         this.cataloged = cataloged;
         this.events = events;

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/binding/config/TestBindingOptionsConfigAdapter.java

@@ -184,6 +184,11 @@ public OptionsConfig adaptFromJson(
                 testOptions.mode(object.getString(MODE_NAME));
             }
 
+            if (object.containsKey(SCHEMA_NAME))
+            {
+                testOptions.schema(object.getString(SCHEMA_NAME));
+            }
+
             if (object.containsKey(CATALOG_NAME))
             {
                 JsonObject catalogsJson = object.getJsonObject(CATALOG_NAME);

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/binding/config/TestBindingOptionsConfigBuilder.java

@@ -31,6 +31,7 @@ public final class TestBindingOptionsConfigBuilder<T> extends ConfigBuilder<T, T
 
     private ModelConfig value;
     private String mode;
+    private String schema;
     private TestAuthorizationConfig authorization;
     private List<CatalogedConfig> catalogs;
     private List<TestBindingOptionsConfig.Event> events;
@@ -64,6 +65,13 @@ public TestBindingOptionsConfigBuilder<T> mode(
         return this;
     }
 
+    public TestBindingOptionsConfigBuilder<T> schema(
+        String schema)
+    {
+        this.schema = schema;
+        return this;
+    }
+
     public TestBindingOptionsConfigBuilder<T> catalog(
         List<CatalogedConfig> catalogs)
     {
@@ -113,7 +121,7 @@ public TestBindingOptionsConfigBuilder<T> vaultAssertion(
     @Override
     public T build()
     {
-        return mapper.apply(new TestBindingOptionsConfig(value, mode, authorization, catalogs, events,
+        return mapper.apply(new TestBindingOptionsConfig(value, mode, schema, authorization, catalogs, events,
                 catalogAssertions, vaultAssertion));
     }
 }

specs/catalog-schema-registry.spec/src/main/scripts/io/aklivity/zilla/specs/catalog/schema/registry/config/register/zilla.yaml

@@ -0,0 +1,51 @@
+#
+# 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
+catalogs:
+  catalog0:
+    type: schema-registry
+    options:
+      url: http://localhost:8081
+bindings:
+  net0:
+    type: test
+    kind: server
+    options:
+      schema: |
+        {
+          "schema":
+          {
+              "type": "record",
+              "name": "test",
+              "fields":
+              [
+                {
+                  "type": "string",
+                  "name": "field1"
+                },
+                {
+                  "type": "com.acme.Referenced",
+                  "name": "int"
+                }
+              ]
+            },
+          "schemaType":"AVRO"
+        }
+      catalog:
+        catalog0:
+          - subject: items-snapshots-value
+    exit: app0

specs/catalog-schema-registry.spec/src/main/scripts/io/aklivity/zilla/specs/catalog/schema/registry/streams/register.schema.rpt

@@ -23,25 +23,25 @@ read http:version "HTTP/1.1"
 read http:header "Host" "localhost:8081"
 read http:header "content-type" "application/json"
 
-read '{'
-       '"schema":"'
-         '{'
-           '\\"type\\": \\"record\\",'
-           '\\"name\\": \\"test\\",'
-           '\\"fields\\":'
-           '['
-             '{'
-               '\\"type\\": \\"string\\",'
-               '\\"name\\": \\"field1\\"'
-             '},'
-             '{'
-               '\\"type\\": \\"com.acme.Referenced\\",'
-               '\\"name\\": \\"int\\"'
-             '}'
-           ']'
-         '}",'
-       '"schemaType":"AVRO"'
-     '}'
+read "{\n"
+     "  \"schema\":\n"
+     "  {\n"
+     "      \"type\": \"record\",\n"
+     "      \"name\": \"test\",\n"
+     "      \"fields\":\n"
+     "      [\n"
+     "        {\n"
+     "          \"type\": \"string\",\n"
+     "          \"name\": \"field1\"\n"
+     "        },\n"
+     "        {\n"
+     "          \"type\": \"com.acme.Referenced\",\n"
+     "          \"name\": \"int\"\n"
+     "        }\n"
+     "      ]\n"
+     "    },\n"
+     "  \"schemaType\":\"AVRO\"\n"
+     "}\n"
 
 read closed
 

specs/engine.spec/src/main/scripts/io/aklivity/zilla/specs/engine/schema/binding/test.schema.patch.json

@@ -38,6 +38,10 @@
                         "type": "object",
                         "properties":
                         {
+                            "schema":
+                            {
+                                "type": "string"
+                            },
                             "value":
                             {
                                 "$ref": "#/$defs/converter"