WIP

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/describe.cluster/cluster.brokers.info/client.rpt

@@ -23,7 +23,6 @@ write zilla:begin.ext ${kafka:beginEx()
                                .request()
                                  .describeCluster()
                                     .includeAuthorizedOperations("false")
-                                    .timeout(0)
                                  .build()
                                .build()}
 

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/describe.cluster/cluster.brokers.info/server.rpt

@@ -27,7 +27,6 @@ read zilla:begin.ext ${kafka:matchBeginEx()
                                .request()
                                  .describeCluster()
                                     .includeAuthorizedOperations("false")
-                                    .timeout(0)
                                  .build()
                                .build()}
 

specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctionsTest.java

@@ -2997,6 +2997,47 @@ public void shouldMatchAlterConfigsResponseBeginExtension() throws Exception
         assertNotNull(matcher.match(byteBuf));
     }
 
+    @Test
+    public void shouldMatchDescribeClusterResponseBeginExtension() throws Exception
+    {
+        BytesMatcher matcher = KafkaFunctions.matchBeginEx()
+                                               .typeId(0x01)
+                                               .response()
+                                                 .describeCluster()
+                                                    .throttle(0)
+                                                    .error((short) 0)
+                                                    .clusterId("cluster-0")
+                                                    .controllerId(0)
+                                                    .broker()
+                                                       .brokerId(1)
+                                                       .host("broker1.example.com")
+                                                       .port(9092)
+                                                       .build()
+                                                    .authorizedOperations(0)
+                                                   .build()
+                                                .build();
+
+        ByteBuffer byteBuf = ByteBuffer.allocate(1024);
+
+        new KafkaBeginExFW.Builder()
+            .wrap(new UnsafeBuffer(byteBuf), 0, byteBuf.capacity())
+            .typeId(0x01)
+            .response(r -> r
+                .describeCluster(d -> d
+                    .throttle(0)
+                    .error((short) 0)
+                    .clusterId("cluster-0")
+                    .controllerId(0)
+                    .brokers(t -> t.item(i -> i
+                        .brokerId(1)
+                        .host("broker1.example.com")
+                        .port(9092)))
+                    .authorizedOperations(0)))
+            .build();
+
+        assertNotNull(matcher.match(byteBuf));
+    }
+
     @Test
     public void shouldMatchFetchBeginExtensionTopic() throws Exception
     {

specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/DescribeClusterIT.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS 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.specs.binding.kafka.streams.application;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.rules.RuleChain.outerRule;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.DisableOnDebug;
+import org.junit.rules.TestRule;
+import org.junit.rules.Timeout;
+
+import io.aklivity.k3po.runtime.junit.annotation.Specification;
+import io.aklivity.k3po.runtime.junit.rules.K3poRule;
+
+public class DescribeClusterIT
+{
+    private final K3poRule k3po = new K3poRule()
+        .addScriptRoot("app", "io/aklivity/zilla/specs/binding/kafka/streams/application/describe.cluster");
+
+    private final TestRule timeout = new DisableOnDebug(new Timeout(5, SECONDS));
+
+    @Rule
+    public final TestRule chain = outerRule(k3po).around(timeout);
+
+    @Test
+    @Specification({
+        "${app}/cluster.brokers.info/client",
+        "${app}/cluster.brokers.info/server"})
+    public void shouldDescribeClusterBrokerInfo() throws Exception
+    {
+        k3po.finish();
+    }
+}

specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/network/DescribeClusterIT.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2021-2023 Aklivity Inc.
+ *
+ * Aklivity licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS 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.specs.binding.kafka.streams.network;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.rules.RuleChain.outerRule;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.DisableOnDebug;
+import org.junit.rules.TestRule;
+import org.junit.rules.Timeout;
+
+import io.aklivity.k3po.runtime.junit.annotation.Specification;
+import io.aklivity.k3po.runtime.junit.rules.K3poRule;
+
+public class DescribeClusterIT
+{
+    private final K3poRule k3po = new K3poRule()
+        .addScriptRoot("net", "io/aklivity/zilla/specs/binding/kafka/streams/network/describe.cluster.v0");
+
+    private final TestRule timeout = new DisableOnDebug(new Timeout(5, SECONDS));
+
+    @Rule
+    public final TestRule chain = outerRule(k3po).around(timeout);
+
+    @Test
+    @Specification({
+        "${net}/cluster.brokers.info/client",
+        "${net}/cluster.brokers.info/server"})
+    public void shouldDescribeClusterBrokerInfo() throws Exception
+    {
+        k3po.finish();
+    }
+}