Add support for describe cluster

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientRequestFactory.java

@@ -57,9 +57,12 @@ public KafkaClientRequestFactory(
     {
         final KafkaClientCreateTopicsFactory clientCreateTopicsFactory = new KafkaClientCreateTopicsFactory(
             config, context, supplyBinding, supplyDebitor, signaler, streamFactory, resolveSasl);
+        final KafkaClientDescribeClusterFactory clientDescribeClusterFactory = new KafkaClientDescribeClusterFactory(
+            config, context, supplyBinding, supplyDebitor, signaler, streamFactory, resolveSasl);
 
         final Int2ObjectHashMap<BindingHandler> factories = new Int2ObjectHashMap<>();
         factories.put(KafkaApi.CREATE_TOPICS.value(), clientCreateTopicsFactory);
+        factories.put(KafkaApi.DESCRIBE_CLUSTER.value(), clientDescribeClusterFactory);
 
         this.kafkaTypeId = context.supplyTypeId(KafkaBinding.NAME);
         this.factories = factories;

runtime/binding-kafka/src/main/zilla/protocol.idl

@@ -598,6 +598,7 @@ scope protocol
 
             struct DescribeClusterResponse
             {
+                int32 correlationId;
                 int32 throttle;
                 int16 error;
                 string16 message;

runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/DescribeClusterIT.java

@@ -0,0 +1,59 @@
+/*
+ * 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.runtime.binding.kafka.internal.stream;
+
+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;
+import io.aklivity.zilla.runtime.engine.test.EngineRule;
+import io.aklivity.zilla.runtime.engine.test.annotation.Configuration;
+
+public class DescribeClusterIT
+{
+    private final K3poRule k3po = new K3poRule()
+        .addScriptRoot("app", "io/aklivity/zilla/specs/binding/kafka/streams/application/describe.cluster")
+        .addScriptRoot("net", "io/aklivity/zilla/specs/binding/kafka/streams/network/describe.cluster.v0");
+
+    private final TestRule timeout = new DisableOnDebug(new Timeout(5, SECONDS));
+
+    private final EngineRule engine = new EngineRule()
+        .directory("target/zilla-itests")
+        .countersBufferCapacity(8192)
+        .configurationRoot("io/aklivity/zilla/specs/binding/kafka/config")
+        .external("net0")
+        .clean();
+
+    @Rule
+    public final TestRule chain = outerRule(engine).around(k3po).around(timeout);
+
+    @Test
+    @Configuration("client.yaml")
+    @Specification({
+        "${app}/cluster.brokers.info/client",
+        "${net}/cluster.brokers.info/server"})
+    public void shouldDescribeClusterBrokerInfo() throws Exception
+    {
+        k3po.finish();
+    }
+}

specs/binding-kafka.spec/src/main/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctions.java

@@ -7572,7 +7572,7 @@ public KafkaDescribeClusterResponseMatcherBuilder build()
                 private boolean match(
                     KafkaBeginExFW beginEx)
                 {
-                    KafkaDescribeClusterResponseBeginExFW describeCluster = beginEx.response().describeCluster();
+                    final KafkaDescribeClusterResponseBeginExFW describeCluster = beginEx.response().describeCluster();
                     return matchThrottle(describeCluster) &&
                         matchError(describeCluster) &&
                         matchMessage(describeCluster) &&

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

@@ -50,5 +50,5 @@ read zilla:begin.ext ${kafka:matchBeginEx()
                                    .build()
                                 .build()}
 
-read closed
 write close
+read closed

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

@@ -53,7 +53,7 @@ write zilla:begin.ext ${kafka:beginEx()
                                    .authorizedOperations(0)
                                    .build()
                                 .build()}
+write flush
 
-write close
 read closed
-
+write close

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

@@ -29,7 +29,7 @@ connect "zilla://streams/net0"
 connected
 
 write 16                                # size
-      32s                               # describe configs
+      60s                               # describe cluster
       0s                                # v0
       ${newRequestId}
       5s "zilla"                        # client id

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

@@ -26,7 +26,7 @@ accepted
 connected
 
 read  16                                # size
-      32s                               # describe configs
+      60s                               # describe cluster
       0s                                # v0
       (int:newRequestId)
       5s "zilla"                        # client id