Checkpoint

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaProxyFactory.java

@@ -65,6 +65,7 @@
 import net.sf.jsqlparser.parser.CCJSqlParserManager;
 import net.sf.jsqlparser.statement.Statement;
 import net.sf.jsqlparser.statement.create.table.CreateTable;
+import net.sf.jsqlparser.statement.drop.Drop;
 
 public final class PgsqlKafkaProxyFactory implements PgsqlKafkaStreamFactory
 {
@@ -1092,7 +1093,7 @@ protected void onKafkaBegin(
             final KafkaBeginExFW kafkaBeginEx =
                 beginEx != null && beginEx.typeId() == kafkaTypeId ? extension.get(kafkaBeginExRO::tryWrap) : null;
 
-            boolean errorExits = kafkaBeginEx.response().createTopics().topics().anyMatch(t -> t.error() != 0);
+            boolean errorExits = kafkaBeginEx.response().deleteTopics().topics().anyMatch(t -> t.error() != 0);
 
             if (!errorExits)
             {
@@ -1384,8 +1385,8 @@ private void decodeDropTopicCommand(
         }
         else if (server.commandsProcessed == 0)
         {
-            final CreateTable createTable = (CreateTable) parseStatement(buffer, offset, length);
-            final String topic = createTable.getTable().getName();
+            final Drop drop = (Drop) parseStatement(buffer, offset, length);
+            final String topic = drop.getName().getName();
 
             final PgsqlKafkaBindingConfig binding = server.binding;
             final String subjectKey = String.format("%s.%s-key", server.database, topic);
@@ -1395,7 +1396,7 @@ else if (server.commandsProcessed == 0)
             binding.catalog.unregister(subjectValue);
 
             final KafkaDeleteTopicsProxy deleteTopicsProxy = server.deleteTopicsProxy;
-            deleteTopicsProxy.doKafkaBegin(traceId, authorization, topics);
+            deleteTopicsProxy.doKafkaBegin(traceId, authorization, List.of("%s.%s".formatted(server.database, topic)));
         }
     }
 

incubator/binding-pgsql-kafka/src/test/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/ProxyIT.java

@@ -58,4 +58,15 @@ public void shouldCreateTopic() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Configuration("proxy.yaml")
+    @Specification({
+        "${pgsql}/drop.topic/client",
+        "${kafka}/drop.topic/server"
+    })
+    public void shouldDropTopic() throws Exception
+    {
+        k3po.finish();
+    }
 }

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

@@ -15,6 +15,7 @@
 package io.aklivity.zilla.runtime.catalog.schema.registry.internal.handler;
 
 import static io.aklivity.zilla.runtime.catalog.schema.registry.internal.handler.CachedSchemaId.IN_PROGRESS;
+import static io.aklivity.zilla.runtime.catalog.schema.registry.internal.serializer.UnregisterSchemaRequest.NO_VERSIONS;
 
 import java.net.URI;
 import java.net.http.HttpClient;
@@ -35,6 +36,7 @@
 import io.aklivity.zilla.runtime.catalog.schema.registry.internal.config.SchemaRegistryCatalogConfig;
 import io.aklivity.zilla.runtime.catalog.schema.registry.internal.events.SchemaRegistryEventContext;
 import io.aklivity.zilla.runtime.catalog.schema.registry.internal.serializer.RegisterSchemaRequest;
+import io.aklivity.zilla.runtime.catalog.schema.registry.internal.serializer.UnregisterSchemaRequest;
 import io.aklivity.zilla.runtime.catalog.schema.registry.internal.types.SchemaRegistryPrefixFW;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.model.function.ValueConsumer;
@@ -56,7 +58,8 @@ public class SchemaRegistryCatalogHandler implements CatalogHandler
 
     private final HttpClient client;
     private final String baseUrl;
-    private final RegisterSchemaRequest request;
+    private final RegisterSchemaRequest registerRequest;
+    private final UnregisterSchemaRequest unregisterRequest;
     private final CRC32C crc32c;
     private final Int2ObjectCache<String> schemas;
     private final Int2ObjectCache<CachedSchemaId> schemaIds;
@@ -71,7 +74,8 @@ public SchemaRegistryCatalogHandler(
     {
         this.baseUrl = catalog.options.url;
         this.client = HttpClient.newHttpClient();
-        this.request = new RegisterSchemaRequest();
+        this.registerRequest = new RegisterSchemaRequest();
+        this.unregisterRequest = new UnregisterSchemaRequest();
         this.crc32c = new CRC32C();
         this.schemas = new Int2ObjectCache<>(1, 1024, i -> {});
         this.schemaIds = new Int2ObjectCache<>(1, 1024, i -> {});
@@ -92,21 +96,25 @@ public int register(
         String response = sendPostHttpRequest(MessageFormat.format(REGISTER_SUBJECT_PATH, subject), schema);
         if (response != null)
         {
-            versionId = request.resolveResponse(response);
+            versionId = registerRequest.resolveResponse(response);
         }
 
         return versionId;
     }
 
     @Override
-    public void unregister(
+    public int[] unregister(
         String subject)
     {
+        int[] versions = NO_VERSIONS;
+
         String response = sendDeleteHttpRequest(MessageFormat.format(UNREGISTER_SUBJECT_PATH, subject));
         if (response != null)
         {
-            request.resolveResponse(response);
+            versions = unregisterRequest.resolveResponse(response);
         }
+
+        return versions;
     }
 
     @Override
@@ -161,7 +169,7 @@ public String resolve(
                             {
                                 event.onRetrievableSchemaId(catalogId, schemaId);
                             }
-                            newFuture.complete(new CachedSchema(request.resolveSchemaResponse(response), retryAttempts));
+                            newFuture.complete(new CachedSchema(registerRequest.resolveSchemaResponse(response), retryAttempts));
                         }
                     }
                     catch (Throwable ex)
@@ -263,8 +271,8 @@ else if (response != null)
                         {
                             event.onRetrievableSchemaSubjectVersion(catalogId, subject, version);
                         }
-                        newFuture.complete(new CachedSchemaId(System.currentTimeMillis(), request.resolveResponse(response),
-                            retryAttempts, retryAfter));
+                        newFuture.complete(new CachedSchemaId(System.currentTimeMillis(),
+                            registerRequest.resolveResponse(response), retryAttempts, retryAfter));
                     }
                 }
                 catch (Throwable ex)

runtime/catalog-schema-registry/src/main/java/io/aklivity/zilla/runtime/catalog/schema/registry/internal/serializer/UnregisterSchemaRequest.java

@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+package io.aklivity.zilla.runtime.catalog.schema.registry.internal.serializer;
+
+import java.io.StringReader;
+import java.util.stream.IntStream;
+
+import jakarta.json.Json;
+import jakarta.json.JsonArray;
+import jakarta.json.JsonReader;
+import jakarta.json.stream.JsonParsingException;
+
+public class UnregisterSchemaRequest
+{
+    public static final int[] NO_VERSIONS = new int[0];
+
+    public int[] resolveResponse(
+        String response)
+    {
+        try
+        {
+            JsonReader reader = Json.createReader(new StringReader(response));
+            JsonArray array = reader.readArray();
+
+            return IntStream.range(0, array.size())
+                .map(array::getInt)
+                .toArray();
+        }
+        catch (JsonParsingException ex)
+        {
+            return NO_VERSIONS;
+        }
+    }
+}

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

@@ -69,9 +69,10 @@ default int register(
         return NO_VERSION_ID;
     }
 
-    default void unregister(
+    default int[] unregister(
         String subject)
     {
+        return new int[0];
     }
 
     String resolve(