Support pgsql-kafka binding

incubator/binding-pgsql-kafka.spec/NOTICE

@@ -14,5 +14,8 @@ This project includes:
   ICU4J under Unicode/ICU License
   Jakarta JSON Processing API under Eclipse Public License 2.0 or GNU General Public License, version 2 with the GNU Classpath Exception
   org.leadpony.justify under The Apache Software License, Version 2.0
+  zilla::incubator::binding-pgsql.spec under Aklivity Community License Agreement
+  zilla::specs::binding-kafka.spec under The Apache Software License, Version 2.0
+  zilla::specs::binding-proxy.spec under The Apache Software License, Version 2.0
   zilla::specs::engine.spec under The Apache Software License, Version 2.0
 

incubator/binding-pgsql-kafka.spec/pom.xml

@@ -39,6 +39,16 @@
       <artifactId>engine.spec</artifactId>
       <version>${project.version}</version>
     </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>binding-pgsql.spec</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+    <dependency>
+      <groupId>${project.groupId}</groupId>
+      <artifactId>binding-kafka.spec</artifactId>
+      <version>${project.version}</version>
+    </dependency>
     <dependency>
       <groupId>junit</groupId>
       <artifactId>junit</artifactId>

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/config/proxy.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: test
+    options:
+      url: http://localhost:8081
+      id: 1
+      schema: |-
+        {
+        "type": "record",
+        "name": "cities",
+        "namespace": "dev",
+        "fields": [
+          {
+            "name": "description",
+            "type": string
+          },
+          {
+            "name": "id",
+            "type": string
+          },
+          {
+            "name": "name",
+            "type": string
+          }
+        ]
+        }
+bindings:
+  app0:
+    type: pgsql-kafka
+    kind: proxy
+    catalog:
+      catalog0:
+        - strategy: topic
+    exit: app1

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/schema/pgsql.kafka.schema.patch.json

@@ -0,0 +1,46 @@
+[
+    {
+        "op": "add",
+        "path": "/$defs/binding/properties/type/enum/-",
+        "value": "pgsql-kafka"
+    },
+    {
+        "op": "add",
+        "path": "/$defs/binding/allOf/-",
+        "value":
+        {
+            "if":
+            {
+               "properties":
+                {
+                    "type":
+                    {
+                        "const": "pgsql-kafka"
+                    }
+                }
+            },
+            "then":
+            {
+                "properties":
+                {
+                    "type":
+                    {
+                        "const": "pgsql-kafka"
+                    },
+                    "kind":
+                    {
+                        "enum": [ "proxy" ]
+                    },
+                    "vault": false,
+                    "options": false,
+                    "routes": false,
+                    "required":
+                    [
+                        "catalog",
+                        "exit"
+                    ]
+                }
+            }
+        }
+    }
+]

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/create.topic/client.rpt

@@ -0,0 +1,85 @@
+#
+# 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.
+#
+
+connect "zilla://streams/app1"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .request()
+                                 .describeCluster()
+                                    .includeAuthorizedOperations("false")
+                                 .build()
+                               .build()}
+
+connected
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                               .typeId(zilla:id("kafka"))
+                               .response()
+                                 .describeCluster()
+                                    .throttle(0)
+                                    .error(0)
+                                    .clusterId("cluster-0")
+                                    .controllerId(0)
+                                    .broker()
+                                       .brokerId(1)
+                                       .host("broker1.example.com")
+                                       .port(9092)
+                                       .build()
+                                   .authorizedOperations(0)
+                                   .build()
+                                .build()}
+
+write close
+read closed
+
+connect "zilla://streams/app1"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .request()
+                                 .createTopics()
+                                    .topic()
+                                       .name("dev.cities")
+                                       .partitionCount(1)
+                                       .replicas(1)
+                                       .assignment(0, 1)
+                                       .config("cleanup.policy", "compact")
+                                       .build()
+                                    .timeout(30000)
+                                    .validateOnly("false")
+                                   .build()
+                                .build()}
+
+connected
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                               .typeId(zilla:id("kafka"))
+                               .response()
+                                 .createTopics()
+                                    .throttle(0)
+                                    .topic()
+                                       .name("dev.cities")
+                                       .error(0)
+                                       .build()
+                                   .build()
+                                .build()}
+
+write close
+read closed

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/create.topic/server.rpt

@@ -0,0 +1,90 @@
+#
+# 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.
+#
+
+property serverAddress "zilla://streams/app1"
+
+accept ${serverAddress}
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+accepted
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                               .typeId(zilla:id("kafka"))
+                               .request()
+                                 .describeCluster()
+                                    .includeAuthorizedOperations("false")
+                                 .build()
+                               .build()}
+
+connected
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .response()
+                                 .describeCluster()
+                                    .throttle(0)
+                                    .error(0)
+                                    .clusterId("cluster-0")
+                                    .controllerId(0)
+                                    .broker()
+                                       .brokerId(1)
+                                       .host("broker1.example.com")
+                                       .port(9092)
+                                       .build()
+                                   .authorizedOperations(0)
+                                   .build()
+                                .build()}
+write flush
+
+read closed
+write close
+
+accepted
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                               .typeId(zilla:id("kafka"))
+                               .request()
+                                 .createTopics()
+                                    .topic()
+                                       .name("dev.cities")
+                                       .partitionCount(1)
+                                       .replicas(1)
+                                       .assignment(0, 1)
+                                       .config("cleanup.policy", "compact")
+                                       .build()
+                                   .timeout(30000)
+                                   .validateOnly("false")
+                                   .build()
+                                .build()}
+
+connected
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .response()
+                                 .createTopics()
+                                    .throttle(0)
+                                    .topic()
+                                       .name("dev.cities")
+                                       .error(0)
+                                       .build()
+                                   .build()
+                                .build()}
+
+write flush
+
+read closed
+write close

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/drop.topic/client.rpt

@@ -0,0 +1,44 @@
+#
+# 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.
+#
+
+connect "zilla://streams/app1"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .request()
+                                 .deleteTopics()
+                                    .topic("dev.cities")
+                                    .timeout(30000)
+                                 .build()
+                               .build()}
+
+connected
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                               .typeId(zilla:id("kafka"))
+                               .response()
+                                 .deleteTopics()
+                                    .throttle(0)
+                                    .topic()
+                                       .name("dev.cities")
+                                       .error(0)
+                                       .build()
+                                   .build()
+                                .build()}
+
+write close
+read closed

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/drop.topic/server.rpt

@@ -0,0 +1,49 @@
+#
+# 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.
+#
+
+property serverAddress "zilla://streams/app1"
+
+accept ${serverAddress}
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+accepted
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                               .typeId(zilla:id("kafka"))
+                               .request()
+                                 .deleteTopics()
+                                    .topic("dev.cities")
+                                    .timeout(30000)
+                                 .build()
+                               .build()}
+
+connected
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .response()
+                                 .deleteTopics()
+                                    .throttle(0)
+                                    .topic()
+                                       .name("dev.cities")
+                                       .error(0)
+                                       .build()
+                                   .build()
+                                .build()}
+write flush
+
+read closed
+write close