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;
+    }
+
     private URI toURI(
         String baseUrl,
         String path)

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

@@ -126,7 +126,7 @@ public void attach(
 
             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 +347,11 @@ else if (assertion.schema != null && !assertion.schema.equals(schema))
                 }
                 else
                 {
-                    if (catalog.subject != null && catalog.version != null)
+                    if (catalog.subject != null && catalog.record != null)
+                    {
+                        handler.register(catalog.subject, catalog.record);
+                    }
+                    else if (catalog.subject != null && catalog.version != null)
                     {
                         handler.resolve(catalog.subject, catalog.version);
                     }

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:
+      catalog:
+        catalog0:
+          - subject: items-snapshots-value
+            record: |
+              {
+                "schema":
+                {
+                    "type": "record",
+                    "name": "test",
+                    "fields":
+                    [
+                      {
+                        "type": "string",
+                        "name": "field1"
+                      },
+                      {
+                        "type": "com.acme.Referenced",
+                        "name": "int"
+                      }
+                    ]
+                  },
+                "schemaType":"AVRO"
+              }
+    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/engine.schema.json

@@ -300,6 +300,26 @@
                         "subject"
                     ],
                     "additionalProperties": false
+                },
+                {
+                    "type": "object",
+                    "properties":
+                    {
+                        "subject":
+                        {
+                            "type": "string"
+                        },
+                        "record":
+                        {
+                            "type": "string",
+                        }
+                    },
+                    "required":
+                    [
+                        "subject",
+                        "record"
+                    ],
+                    "additionalProperties": false
                 }
             ]
         },