Fix message fragmentation in sse-kafka and flow control in kafka merge

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

@@ -1334,9 +1334,7 @@ private void onMergedReplyWindow(
             assert acknowledge >= replyAck;
             assert maximum >= replyMax;
 
-            final int replyNoAck = (int)(replySeq - replyAck);
-            final int newReplyNoAck = (int)(sequence - acknowledge);
-            final int credit = (replyNoAck - newReplyNoAck) + (maximum - replyMax);
+            final int credit = (int)(acknowledge - replyAck) + (maximum - replyMax);
             assert credit >= 0;
 
             this.replyAck = acknowledge;
@@ -1619,9 +1617,9 @@ private void doMergedReplyEndIfNecessary(
         }
 
         private void doMergedReplyFlush(
-                long traceId,
-                int reserved,
-                KafkaFlushExFW kafkaFlushEx)
+            long traceId,
+            int reserved,
+            KafkaFlushExFW kafkaFlushEx)
         {
             final KafkaFetchFlushExFW kafkaFetchFlushEx = kafkaFlushEx.fetch();
             kafkaFetchFlushEx.partition().partitionId();

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

@@ -55,6 +55,7 @@
 
 public final class SseKafkaProxyFactory implements SseKafkaStreamFactory
 {
+    private static final int INIT_FLAG = 0x02;
     private static final String SSE_TYPE_NAME = "sse";
     private static final String KAFKA_TYPE_NAME = "kafka";
 
@@ -643,28 +644,33 @@ private void onKafkaData(
             }
             else
             {
+                String8FW encodedId = null;
+                OctetsFW key = null;
                 final int flags = data.flags();
                 final OctetsFW payload = data.payload();
-                final OctetsFW extension = data.extension();
-                final ExtensionFW dataEx = extension.get(extensionRO::tryWrap);
-                final KafkaDataExFW kafkaDataEx =
-                        dataEx != null && dataEx.typeId() == kafkaTypeId ? extension.get(kafkaDataExRO::tryWrap) : null;
-                final KafkaMergedDataExFW kafkaMergedDataEx =
-                        kafkaDataEx != null && kafkaDataEx.kind() == KafkaDataExFW.KIND_MERGED ? kafkaDataEx.merged() : null;
-                final Array32FW<KafkaOffsetFW> progress = kafkaMergedDataEx != null ? kafkaMergedDataEx.progress() : null;
-                final OctetsFW key = kafkaMergedDataEx != null ? kafkaMergedDataEx.key().value() : null;
-                final Array32FW<KafkaHeaderFW> headers = kafkaMergedDataEx != null ? kafkaMergedDataEx.headers() : null;
-                final KafkaHeaderFW etag = headers.matchFirst(h -> HEADER_NAME_ETAG.value().equals(h.name().value()));
 
-                String8FW encodedId = null;
-                switch (delegate.resolved.eventId())
+                if ((flags & INIT_FLAG) != 0x00)
                 {
-                case EVENT_ID_KEY64_AND_ETAG:
-                    encodedId = sseEventId.encodeKeyAndProgress(key, progress, etag);
-                    break;
-                case EVENT_ID_ETAG_ONLY:
-                    encodedId = sseEventId.encodeProgressOnly(progress, etag);
-                    break;
+                    final OctetsFW extension = data.extension();
+                    final ExtensionFW dataEx = extension.get(extensionRO::tryWrap);
+                    final KafkaDataExFW kafkaDataEx =
+                        dataEx != null && dataEx.typeId() == kafkaTypeId ? extension.get(kafkaDataExRO::tryWrap) : null;
+                    final KafkaMergedDataExFW kafkaMergedDataEx =
+                        kafkaDataEx != null && kafkaDataEx.kind() == KafkaDataExFW.KIND_MERGED ? kafkaDataEx.merged() : null;
+                    final Array32FW<KafkaOffsetFW> progress = kafkaMergedDataEx != null ? kafkaMergedDataEx.progress() : null;
+                    key = kafkaMergedDataEx != null ? kafkaMergedDataEx.key().value() : null;
+                    final Array32FW<KafkaHeaderFW> headers = kafkaMergedDataEx != null ? kafkaMergedDataEx.headers() : null;
+                    final KafkaHeaderFW etag = headers.matchFirst(h -> HEADER_NAME_ETAG.value().equals(h.name().value()));
+
+                    switch (delegate.resolved.eventId())
+                    {
+                    case EVENT_ID_KEY64_AND_ETAG:
+                        encodedId = sseEventId.encodeKeyAndProgress(key, progress, etag);
+                        break;
+                    case EVENT_ID_ETAG_ONLY:
+                        encodedId = sseEventId.encodeProgressOnly(progress, etag);
+                        break;
+                    }
                 }
 
                 final String8FW eventType = payload == null ? EVENT_TYPE_DELETE : EVENT_TYPE_MESSAGE;

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

@@ -157,6 +157,16 @@ public void shouldReceiveServerSentMessagesWithEtag() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("proxy.with.topic.yaml")
+    @Specification({
+        "${sse}/server.sent.100k.message/client",
+        "${kafka}/server.sent.100k.message/server"})
+    public void shouldReceiveServerSent100kMessage() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("proxy.with.topic.and.event.id.yaml")
     @Specification({

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/kafka/server.sent.100k.message/client.rpt

@@ -0,0 +1,41 @@
+#
+# 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/kafka0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                                .capabilities("FETCH_ONLY")
+                                .topic("test")
+                                .partition(-1, -2)
+                                .build()
+                            .build()}
+
+connected
+
+read zilla:data.ext ${kafka:matchDataEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                                .partition(0, 1, 2)
+                                .progress(0, 2)
+                                .progress(1, 1)
+                                .key("key")
+                                .header("header", "value")
+                                .build()
+                            .build()}
+read [0...100000]

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/kafka/server.sent.100k.message/server.rpt

@@ -0,0 +1,48 @@
+#
+# 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 data ${http:randomBytes(100000)}
+
+accept "zilla://streams/kafka0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+accepted
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                                .capabilities("FETCH_ONLY")
+                                .topic("test")
+                                .partition(-1, -2)
+                                .build()
+                            .build()}
+
+connected
+
+write zilla:data.ext ${kafka:dataEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                                .timestamp(kafka:timestamp())
+                                .partition(0, 1, 2)
+                                .progress(0, 2)
+                                .progress(1, 1)
+                                .key("key")
+                                .header("header", "value")
+                                .build()
+                            .build()}
+write ${data}
+write flush
+

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/sse/server.sent.100k.message/client.rpt

@@ -0,0 +1,35 @@
+#
+# 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/sse0"
+        option zilla:window 8192
+        option zilla:transmission "duplex"
+
+write zilla:begin.ext ${sse:beginEx()
+                           .typeId(zilla:id("sse"))
+                           .scheme("http")
+                           .authority("localhost:8080")
+                           .path("/test")
+                           .lastId(null)
+                           .build()}
+
+connected
+
+read zilla:data.ext ${sse:matchDataEx()
+                         .typeId(zilla:id("sse"))
+                         .id("AQQABAIC")
+                         .type(null)
+                         .build()}
+read [0..100000]

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/sse/server.sent.100k.message/server.rpt

@@ -0,0 +1,40 @@
+#
+# 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 data ${http:randomBytes(100000)}
+
+accept "zilla://streams/sse0"
+       option zilla:window 8192
+       option zilla:transmission "duplex"
+accepted
+
+read zilla:begin.ext ${sse:beginEx()
+                           .typeId(zilla:id("sse"))
+                           .scheme("http")
+                           .authority("localhost:8080")
+                           .path("/test")
+                           .lastId(null)
+                           .build()}
+
+connected
+
+write zilla:data.ext ${sse:dataEx()
+                         .typeId(zilla:id("sse"))
+                         .id("AQQABAIC")
+                         .type(null)
+                         .build()}
+write ${data}
+write flush
+

specs/binding-sse-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/sse/kafka/streams/KafkaIT.java

@@ -151,4 +151,13 @@ public void shouldReceiveClientSentAbort() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Specification({
+        "${kafka}/server.sent.100k.message/client",
+        "${kafka}/server.sent.100k.message/server"})
+    public void shouldReceiveServerSent100kMessage() throws Exception
+    {
+        k3po.finish();
+    }
 }

specs/binding-sse-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/sse/kafka/streams/SseIT.java

@@ -178,4 +178,13 @@ public void shouldRejectClientSentMessage() throws Exception
     {
         k3po.finish();
     }
+
+    @Test
+    @Specification({
+        "${sse}/server.sent.100k.message/client",
+        "${sse}/server.sent.100k.message/server"})
+    public void shouldReceiveServerSent100kMessage() throws Exception
+    {
+        k3po.finish();
+    }
 }