Merge branch 'release/0.9.97'

aklivity · Oct 7, 2024 · 3f67e32 · 3f67e32
2 parents e70a261 + 91dfe54
commit 3f67e32
Show file tree

Hide file tree

Showing 178 changed files with 2,269 additions and 359 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,7 +2,23 @@
 
 ## [Unreleased](https://github.com/aklivity/zilla/tree/HEAD)
 
-[Full Changelog](https://github.com/aklivity/zilla/compare/0.9.95...HEAD)
+[Full Changelog](https://github.com/aklivity/zilla/compare/0.9.96...HEAD)
+
+**Implemented enhancements:**
+
+- Support `jwt` guarded identity via custom token claim [\#1276](https://github.com/aklivity/zilla/issues/1276)
+- Support `insert into` to seed `kafka` messages via `risingwave` binding [\#1274](https://github.com/aklivity/zilla/issues/1274)
+
+**Merged pull requests:**
+
+- `pgsql` DROP TOPIC command to KafkaDeleteTopicsBeginEx plus catalog unregister subject [\#1280](https://github.com/aklivity/zilla/pull/1280) ([akrambek](https://github.com/akrambek))
+- external udf - python support [\#1278](https://github.com/aklivity/zilla/pull/1278) ([ankitk-me](https://github.com/ankitk-me))
+- Support jwt guarded identity via custom token claim [\#1277](https://github.com/aklivity/zilla/pull/1277) ([akrambek](https://github.com/akrambek))
+- Support insert into to seed kafka messages via risingwave binding [\#1275](https://github.com/aklivity/zilla/pull/1275) ([akrambek](https://github.com/akrambek))
+
+## [0.9.96](https://github.com/aklivity/zilla/tree/0.9.96) (2024-10-01)
+
+[Full Changelog](https://github.com/aklivity/zilla/compare/0.9.95...0.9.96)
 
 **Implemented enhancements:**
 

diff --git a/build/flyweight-maven-plugin/pom.xml b/build/flyweight-maven-plugin/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>build</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/build/pom.xml b/build/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>zilla</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/cloud/docker-image/pom.xml b/cloud/docker-image/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>cloud</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/cloud/helm-chart/pom.xml b/cloud/helm-chart/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>cloud</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/cloud/pom.xml b/cloud/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>zilla</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/conf/pom.xml b/conf/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>zilla</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/incubator/binding-amqp.spec/pom.xml b/incubator/binding-amqp.spec/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>incubator</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/incubator/binding-amqp/pom.xml b/incubator/binding-amqp/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>incubator</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/incubator/binding-pgsql-kafka.spec/pom.xml b/incubator/binding-pgsql-kafka.spec/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>incubator</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/incubator/binding-pgsql-kafka/pom.xml b/incubator/binding-pgsql-kafka/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>incubator</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/.../io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaCommandType.java b/.../io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaCommandType.java
@@ -22,6 +22,7 @@
 public enum PgsqlKafkaCommandType
 {
     CREATE_TOPIC_COMMAND("CREATE TOPIC".getBytes()),
+    DROP_TOPIC_COMMAND("DROP TOPIC".getBytes()),
     UNKNOWN_COMMAND("UNKNOWN".getBytes());
 
     private final byte[] value;

diff --git a/...livity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaCompletionCommand.java b/...livity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaCompletionCommand.java
@@ -17,6 +17,7 @@
 public enum PgsqlKafkaCompletionCommand
 {
     CREATE_TOPIC_COMMAND("CREATE_TOPIC".getBytes()),
+    DROP_TOPIC_COMMAND("DROP_TOPIC".getBytes()),
     UNKNOWN_COMMAND("UNKNOWN".getBytes());
 
     private final byte[] value;

diff --git a/...io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaProxyFactory.java b/...io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/PgsqlKafkaProxyFactory.java
@@ -16,6 +16,7 @@
 
 import static io.aklivity.zilla.runtime.engine.buffer.BufferPool.NO_SLOT;
 import static io.aklivity.zilla.runtime.engine.catalog.CatalogHandler.NO_VERSION_ID;
+import static java.nio.charset.StandardCharsets.UTF_8;
 import static java.util.Objects.requireNonNull;
 
 import java.io.InputStreamReader;
@@ -64,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
 {
@@ -143,6 +145,7 @@ public final class PgsqlKafkaProxyFactory implements PgsqlKafkaStreamFactory
         Object2ObjectHashMap<PgsqlKafkaCommandType, PgsqlDecoder> pgsqlDecoder =
             new Object2ObjectHashMap<>();
         pgsqlDecoder.put(PgsqlKafkaCommandType.CREATE_TOPIC_COMMAND, this::decodeCreateTopicCommand);
+        pgsqlDecoder.put(PgsqlKafkaCommandType.DROP_TOPIC_COMMAND, this::decodeDropTopicCommand);
         pgsqlDecoder.put(PgsqlKafkaCommandType.UNKNOWN_COMMAND, this::decodeUnknownCommand);
         this.pgsqlDecoder = pgsqlDecoder;
     }
@@ -232,6 +235,7 @@ private final class PgsqlProxy
         private final String database;
         private final PgsqlKafkaBindingConfig binding;
         private final KafkaCreateTopicsProxy createTopicsProxy;
+        private final KafkaDeleteTopicsProxy deleteTopicsProxy;
 
         private final IntArrayQueue queries;
 
@@ -281,6 +285,7 @@ private PgsqlProxy(
             this.queries = new IntArrayQueue();
 
             this.createTopicsProxy = new KafkaCreateTopicsProxy(routedId, resolvedId, this);
+            this.deleteTopicsProxy = new KafkaDeleteTopicsProxy(routedId, resolvedId, this);
         }
 
         private void onAppMessage(
@@ -488,7 +493,7 @@ private void onCommandCompleted(
             doAppWindow(traceId, authorization);
         }
 
-        public void onKafkaCreateTopicsBegin(
+        public void onKafkaBegin(
             long traceId,
             long authorization)
         {
@@ -1018,7 +1023,81 @@ protected void onKafkaBegin(
 
             if (!errorExits)
             {
-                delegate.onKafkaCreateTopicsBegin(traceId, authorization);
+                delegate.onKafkaBegin(traceId, authorization);
+
+                doKafkaWindow(traceId, authorization);
+                doKafkaEnd(traceId, authorization);
+            }
+            else
+            {
+                delegate.cleanup(traceId, authorization);
+            }
+        }
+    }
+
+    private final class KafkaDeleteTopicsProxy extends KafkaProxy
+    {
+        private KafkaDeleteTopicsProxy(
+            long originId,
+            long routedId,
+            PgsqlProxy delegate)
+        {
+            super(originId, routedId, delegate);
+        }
+
+        private void doKafkaBegin(
+            long traceId,
+            long authorization,
+            List<String> topics)
+        {
+            initialSeq = delegate.initialSeq;
+            initialAck = delegate.initialAck;
+            initialMax = delegate.initialMax;
+            state = PgsqlKafkaState.openingInitial(state);
+
+            final KafkaBeginExFW kafkaBeginEx =
+                    kafkaBeginExRW.wrap(extBuffer, 0, extBuffer.capacity())
+                        .typeId(kafkaTypeId)
+                        .request(r -> r
+                            .deleteTopics(c -> c
+                                .names(ct ->
+                                    topics.forEach(t -> ct.item(i -> i.set(t, UTF_8))))
+                            .timeout(config.kafkaTopicRequestTimeoutMs())))
+                        .build();
+
+            kafka = newKafkaConsumer(this::onKafkaMessage, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                traceId, authorization, 0, kafkaBeginEx);
+        }
+
+        @Override
+        protected void onKafkaBegin(
+            BeginFW begin)
+        {
+            final long sequence = begin.sequence();
+            final long acknowledge = begin.acknowledge();
+            final long traceId = begin.traceId();
+            final long authorization = begin.authorization();
+            final OctetsFW extension = begin.extension();
+
+            assert acknowledge <= sequence;
+            assert sequence >= replySeq;
+            assert acknowledge >= replyAck;
+
+            replySeq = sequence;
+            replyAck = acknowledge;
+            state = PgsqlKafkaState.openingReply(state);
+
+            assert replyAck <= replySeq;
+
+            final ExtensionFW beginEx = extension.get(extensionRO::tryWrap);
+            final KafkaBeginExFW kafkaBeginEx =
+                beginEx != null && beginEx.typeId() == kafkaTypeId ? extension.get(kafkaBeginExRO::tryWrap) : null;
+
+            boolean errorExits = kafkaBeginEx.response().deleteTopics().topics().anyMatch(t -> t.error() != 0);
+
+            if (!errorExits)
+            {
+                delegate.onKafkaBegin(traceId, authorization);
 
                 doKafkaWindow(traceId, authorization);
                 doKafkaEnd(traceId, authorization);
@@ -1292,6 +1371,35 @@ else if (server.commandsProcessed == 0)
         }
     }
 
+    private void decodeDropTopicCommand(
+        PgsqlProxy server,
+        long traceId,
+        long authorization,
+        DirectBuffer buffer,
+        int offset,
+        int length)
+    {
+        if (server.commandsProcessed == 1)
+        {
+            server.onCommandCompleted(traceId, authorization, length, PgsqlKafkaCompletionCommand.DROP_TOPIC_COMMAND);
+        }
+        else if (server.commandsProcessed == 0)
+        {
+            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);
+            final String subjectValue = String.format("%s.%s-value", server.database, topic);
+
+            binding.catalog.unregister(subjectKey);
+            binding.catalog.unregister(subjectValue);
+
+            final KafkaDeleteTopicsProxy deleteTopicsProxy = server.deleteTopicsProxy;
+            deleteTopicsProxy.doKafkaBegin(traceId, authorization, List.of("%s.%s".formatted(server.database, topic)));
+        }
+    }
+
     private void decodeUnknownCommand(
         PgsqlProxy server,
         long traceId,
@@ -1351,6 +1459,13 @@ private Statement parseStatement(
                 sql = sql.replace("CREATE TOPIC", "CREATE TABLE");
                 statement = parserManager.parse(new StringReader(sql));
             }
+            if (decodeCommandType(buffer, offset, length).
+                    equals(PgsqlKafkaCommandType.DROP_TOPIC_COMMAND))
+            {
+                String sql = buffer.getStringWithoutLengthUtf8(offset, length);
+                sql = sql.replace("DROP TOPIC", "DROP TABLE");
+                statement = parserManager.parse(new StringReader(sql));
+            }
             else
             {
                 inputStream.wrap(buffer, offset, length);

diff --git a/.../src/test/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/ProxyIT.java b/.../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();
+    }
 }
diff --git a/incubator/binding-pgsql.spec/pom.xml b/incubator/binding-pgsql.spec/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>incubator</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/incubator/binding-pgsql/pom.xml b/incubator/binding-pgsql/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>incubator</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/incubator/binding-risingwave.spec/pom.xml b/incubator/binding-risingwave.spec/pom.xml
@@ -8,7 +8,7 @@
   <parent>
     <groupId>io.aklivity.zilla</groupId>
     <artifactId>incubator</artifactId>
-    <version>0.9.96</version>
+    <version>0.9.97</version>
     <relativePath>../pom.xml</relativePath>
   </parent>
 

diff --git a/...ec/src/main/scripts/io/aklivity/zilla/specs/binding/risingwave/config/proxy.function.yaml b/...ec/src/main/scripts/io/aklivity/zilla/specs/binding/risingwave/config/proxy.function.yaml
@@ -22,4 +22,6 @@ bindings:
     options:
       udf:
         - server: http://localhost:8815
+        - server: http://localhost:8816
+          language: python
     exit: app1
diff --git a/...in/scripts/io/aklivity/zilla/specs/binding/risingwave/schema/risingwave.schema.patch.json b/...in/scripts/io/aklivity/zilla/specs/binding/risingwave/schema/risingwave.schema.patch.json
@@ -80,13 +80,16 @@
                                             "title": "Language",
                                             "type": "string",
                                             "default": "java",
-                                            "const": "java"
+                                            "enum":
+                                            [
+                                                "java",
+                                                "python"
+                                            ]
                                         }
                                     },
                                     "additionalProperties": false
                                 },
-                                "minItems": 1,
-                                "maxItems": 1
+                                "minItems": 1
                             }
                         },
                         "additionalProperties": false