Add more spec

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

@@ -0,0 +1,79 @@
+/*
+ * 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 io.aklivity.zilla.runtime.binding.kafka.internal.KafkaConfigurationTest.KAFKA_CLIENT_SASL_SCRAM_NONCE_NAME;
+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 org.kaazing.k3po.junit.annotation.Specification;
+import org.kaazing.k3po.junit.rules.K3poRule;
+
+import io.aklivity.zilla.runtime.engine.test.EngineRule;
+import io.aklivity.zilla.runtime.engine.test.annotation.Configuration;
+import io.aklivity.zilla.runtime.engine.test.annotation.Configure;
+
+public class ClientInitProducerIdSaslIT
+{
+    private final K3poRule k3po = new K3poRule()
+        .addScriptRoot("net", "io/aklivity/zilla/specs/binding/kafka/streams/network/init.producer.id.v4.sasl.handshake.v1")
+        .addScriptRoot("app", "io/aklivity/zilla/specs/binding/kafka/streams/application/init.producer.id");
+
+    private final TestRule timeout = new DisableOnDebug(new Timeout(15, 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.options.sasl.plain.yaml")
+    @Specification({
+        "${app}/produce.new.id/client",
+        "${net}/produce.new.id.sasl.plain/server"})
+    public void shouldGenerateNewProducerIdWithSaslPlain() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Configuration("client.options.sasl.scram.yaml")
+    @Specification({
+        "${app}/produce.new.id/client",
+        "${net}/produce.new.id.sasl.scram/server"})
+    @Configure(name = KAFKA_CLIENT_SASL_SCRAM_NONCE_NAME,
+        value = "io.aklivity.zilla.runtime.binding.kafka.internal.stream.ClientInitProducerIdSaslIT::supplyNonce")
+    public void shouldGenerateNewProducerIdWithSaslScram() throws Exception
+    {
+        k3po.finish();
+    }
+
+    public static String supplyNonce()
+    {
+        return "fyko+d2lbbFgONRv9qkxdawL";
+    }
+}

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/init.producer.id/produce.new.id/server.rpt

@@ -41,3 +41,4 @@ write zilla:begin.ext ${kafka:beginEx()
                                 .producerEpoch(2)
                                 .build()
                             .build()}
+write flush

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/merged/merged.produce.message.values.producer.id/client.rpt

@@ -0,0 +1,111 @@
+#
+# 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.
+#
+
+property deltaMillis 0L
+property newTimestamp ${kafka:timestamp() + deltaMillis}
+
+connect "zilla://streams/app0"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .merged()
+                                   .capabilities("PRODUCE_ONLY")
+                                   .topic("test")
+                                   .ackMode("LEADER_ONLY")
+                                   .build()
+                               .build()}
+
+connected
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                .produce()
+                                  .timestamp(newTimestamp)
+                                   .producerId(1)
+                                   .producerEpoch(1)
+                                  .partition(0, 1)
+                                  .build()
+                              .build()}
+write "Hello, world #A1"
+write flush
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                .produce()
+                                  .timestamp(newTimestamp)
+                                  .producerId(1)
+                                  .producerEpoch(1)
+                                  .partition(1, 1)
+                                  .build()
+                              .build()}
+write "Hello, world #B1"
+write flush
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                .produce()
+                                  .timestamp(newTimestamp)
+                                  .producerId(1)
+                                  .producerEpoch(1)
+                                  .partition(0, 2)
+                                  .build()
+                              .build()}
+write "Hello, world #A2"
+write flush
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                .produce()
+                                  .timestamp(newTimestamp)
+                                  .producerId(1)
+                                  .producerEpoch(1)
+                                  .partition(1, 2)
+                                  .build()
+                              .build()}
+write "Hello, world #B2"
+write flush
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                .produce()
+                                  .timestamp(newTimestamp)
+                                  .producerId(1)
+                                  .producerEpoch(1)
+                                  .partition(2, 1)
+                                  .build()
+                              .build()}
+write "Hello, world #C1"
+write flush
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                .produce()
+                                  .timestamp(newTimestamp)
+                                  .producerId(1)
+                                  .producerEpoch(1)
+                                  .partition(2, 2)
+                                  .build()
+                              .build()}
+write "Hello, world #C2"
+write flush

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/merged/merged.produce.message.values.producer.id/server.rpt

@@ -0,0 +1,98 @@
+#
+# 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.
+#
+
+accept "zilla://streams/app0"
+    option zilla:window 16
+    option zilla:transmission "half-duplex"
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                  .capabilities("PRODUCE_ONLY")
+                                  .topic("test")
+                                  .ackMode("LEADER_ONLY")
+                                  .build()
+                              .build()}
+
+connected
+
+read zilla:data.ext ${kafka:matchDataEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                .produce()
+                                 .producerId(1)
+                                 .producerEpoch(1)
+                                 .partition(0, 1)
+                                 .build()
+                             .build()}
+read "Hello, world #A1"
+
+read zilla:data.ext ${kafka:matchDataEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                .produce()
+                                 .producerId(1)
+                                 .producerEpoch(1)
+                                 .partition(1, 1)
+                                 .build()
+                             .build()}
+read "Hello, world #B1"
+
+read zilla:data.ext ${kafka:matchDataEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                .produce()
+                                 .producerId(1)
+                                 .producerEpoch(1)
+                                 .partition(0, 2)
+                                 .build()
+                             .build()}
+read "Hello, world #A2"
+
+read zilla:data.ext ${kafka:matchDataEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                .produce()
+                                 .producerId(1)
+                                 .producerEpoch(1)
+                                 .partition(1, 2)
+                                 .build()
+                             .build()}
+read "Hello, world #B2"
+
+read zilla:data.ext ${kafka:matchDataEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                .produce()
+                                 .producerId(1)
+                                 .producerEpoch(1)
+                                 .partition(2, 1)
+                                 .build()
+                             .build()}
+read "Hello, world #C1"
+
+read zilla:data.ext ${kafka:matchDataEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                .produce()
+                                 .producerId(1)
+                                 .producerEpoch(1)
+                                 .partition(2, 2)
+                                 .build()
+                             .build()}
+read "Hello, world #C2"