Fix large message hashing in http-kafka

runtime/binding-http-kafka/src/main/java/io/aklivity/zilla/runtime/binding/http/kafka/internal/config/HttpKafkaWithProduceHash.java

@@ -44,8 +44,18 @@ public class HttpKafkaWithProduceHash
     public void updateHash(
         DirectBuffer value)
     {
-        value.getBytes(0, hashBytesRW, 0, value.capacity());
-        md5.update(hashBytesRW, 0, value.capacity());
+        final int chunkSize = hashBytesRW.length;
+        int offset = 0;
+        int remaining = value.capacity();
+
+        while (remaining > 0)
+        {
+            int bytesToRead = Math.min(chunkSize, remaining);
+            value.getBytes(offset, hashBytesRW, 0, bytesToRead);
+            md5.update(hashBytesRW, 0, bytesToRead);
+            offset += bytesToRead;
+            remaining -= bytesToRead;
+        }
     }
 
     public void digestHash()

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

@@ -40,7 +40,7 @@ public class HttpKafkaProxyIT
     private final EngineRule engine = new EngineRule()
         .directory("target/zilla-itests")
         .countersBufferCapacity(8192)
-        .configure(ENGINE_BUFFER_SLOT_CAPACITY, 8192)
+        .configure(ENGINE_BUFFER_SLOT_CAPACITY, 16384)
         .configurationRoot("io/aklivity/zilla/specs/binding/http/kafka/config")
         .external("kafka0")
         .clean();
@@ -598,6 +598,16 @@ public void shouldPostItemCommand() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("proxy.post.item.command.yaml")
+    @Specification({
+        "${http}/post.item.command.10k/client",
+        "${kafka}/post.item.command.10k/server"})
+    public void shouldPostItemCommand10k() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("proxy.post.item.command.yaml")
     @Specification({

specs/binding-http-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/http/kafka/streams/http/post.item.command.10k/client.rpt

@@ -0,0 +1,43 @@
+#
+# Copyright 2021-2024 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/http0"
+        option zilla:window 16384
+        option zilla:transmission "half-duplex"
+        option zilla:update "proactive"
+
+write zilla:begin.ext ${http:beginEx()
+                            .typeId(zilla:id("http"))
+                            .header(":method", "POST")
+                            .header(":scheme", "https")
+                            .header(":authority", "example.com:9090")
+                            .header(":path", "/items/92d0bf92-63e0-4cfc-ae73-71dee92d1544/rename")
+                            .header("content-type", "application/json")
+                            .header("content-length", "10000")
+                            .header("idempotency-key", "59410e57-3e0f-4b61-9328-f645a7968ac8")
+                            .build()}
+
+connected
+
+write ${http:constantBytes(10000)}
+
+write close
+
+read zilla:begin.ext ${http:matchBeginEx()
+                           .typeId(zilla:id("http"))
+                           .header(":status", "204")
+                           .build()}
+
+read closed

specs/binding-http-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/http/kafka/streams/http/post.item.command.10k/server.rpt

@@ -0,0 +1,43 @@
+#
+# Copyright 2021-2024 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.
+#
+
+accept "zilla://streams/http0"
+       option zilla:window 16384
+       option zilla:transmission "half-duplex"
+accepted
+
+read zilla:begin.ext ${http:matchBeginEx()
+                           .typeId(zilla:id("http"))
+                           .header(":method", "POST")
+                           .header(":path", "/items/92d0bf92-63e0-4cfc-ae73-71dee92d1544/rename")
+                           .header("content-type", "application/json")
+                           .header("content-length", "10000")
+                           .header("idempotency-key", "59410e57-3e0f-4b61-9328-f645a7968ac8")
+                           .build()}
+
+connected
+
+read [0..10000]
+
+read closed
+
+write zilla:begin.ext ${http:beginEx()
+                            .typeId(zilla:id("http"))
+                            .header(":status", "204")
+                            .build()}
+
+write flush
+
+write close

specs/binding-http-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/http/kafka/streams/kafka/post.item.command.10k/client.rpt

@@ -0,0 +1,117 @@
+#
+# Copyright 2021-2024 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 16384
+    option zilla:transmission "duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                 .capabilities("PRODUCE_ONLY")
+                                 .topic("items-commands")
+                                 .partition(-1, -2)
+                                 .ackMode("LEADER_ONLY")
+                                 .build()
+                             .build()}
+
+connected
+
+write option zilla:flags "init"
+write zilla:data.ext ${kafka:dataEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                               .produce()
+                                .deferred(10000)
+                                .partition(-1, -1)
+                                .key("92d0bf92-63e0-4cfc-ae73-71dee92d1544")
+                                .header(":method", "POST")
+                                .header(":scheme", "https")
+                                .header(":authority", "example.com:9090")
+                                .header(":path", "/items/92d0bf92-63e0-4cfc-ae73-71dee92d1544/rename")
+                                .header("content-type", "application/json")
+                                .header("idempotency-key", "59410e57-3e0f-4b61-9328-f645a7968ac8")
+                                .header("zilla:reply-to", "items-replies")
+                                .header("zilla:identity", "test")
+                                .header("zilla:command", "rename")
+                                .build()
+                            .build()}
+write zilla:data.empty
+write flush
+
+write option zilla:flags "none"
+write ${kafka:randomBytes(10000)}
+write flush
+
+write notify SENDING_ASYNC_REQUEST
+
+write await RECEIVED_ASYNC_FLUSH
+
+write option zilla:flags "fin"
+write zilla:data.ext ${kafka:dataEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                               .produce()
+                                .partition(-1, -1)
+                                .header("zilla:correlation-id", "59410e57-3e0f-4b61-9328-f645a7968ac8-55e623a5b35ce648e9af4ff07f8d307b")
+                                .build()
+                            .build()}
+write zilla:data.empty
+write flush
+
+write close
+read closed
+
+connect await SENDING_ASYNC_REQUEST
+        "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("items-replies")
+                                   .partition(-1, -2)
+                                   .filter()
+                                     .header("zilla:correlation-id", "59410e57-3e0f-4b61-9328-f645a7968ac8-55e623a5b35ce648e9af4ff07f8d307b")
+                                     .build()
+                                   .build()
+                               .build()}
+
+connected
+
+read advised zilla:flush
+
+read notify RECEIVED_ASYNC_FLUSH
+
+read zilla:data.ext ${kafka:matchDataEx()
+                           .typeId(zilla:id("kafka"))
+                           .merged()
+                             .fetch()
+                               .partition(0, 1, 2)
+                               .progress(0, 2)
+                               .progress(1, 1)
+                               .key("92d0bf92-63e0-4cfc-ae73-71dee92d1544")
+                               .header(":status", "204")
+                               .header("zilla:correlation-id", "59410e57-3e0f-4b61-9328-f645a7968ac8-55e623a5b35ce648e9af4ff07f8d307b")
+                               .build()
+                           .build()}
+read zilla:data.null
+
+read closed
+write close
+
+

specs/binding-http-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/http/kafka/streams/kafka/post.item.command.10k/server.rpt

@@ -0,0 +1,113 @@
+#
+# Copyright 2021-2024 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.
+#
+
+accept "zilla://streams/kafka0"
+    option zilla:window 16384
+    option zilla:transmission "duplex"
+
+accepted
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                                .capabilities("PRODUCE_ONLY")
+                                .topic("items-commands")
+                                .partition(-1, -2)
+                                .ackMode("LEADER_ONLY")
+                                .build()
+                            .build()}
+
+connected
+
+read option zilla:flags "init"
+read zilla:data.ext ${kafka:matchDataEx()
+                           .typeId(zilla:id("kafka"))
+                           .merged()
+                             .produce()
+                               .deferred(10000)
+                               .partition(-1, -1)
+                               .key("92d0bf92-63e0-4cfc-ae73-71dee92d1544")
+                               .header(":method", "POST")
+                               .header(":scheme", "https")
+                               .header(":authority", "example.com:9090")
+                               .header(":path", "/items/92d0bf92-63e0-4cfc-ae73-71dee92d1544/rename")
+                               .header("content-type", "application/json")
+                               .header("idempotency-key", "59410e57-3e0f-4b61-9328-f645a7968ac8")
+                               .header("zilla:reply-to", "items-replies")
+                               .header("zilla:identity", "test")
+                               .header("zilla:command", "rename")
+                               .build()
+                           .build()}
+read zilla:data.empty
+
+read option zilla:flags "none"
+read [0..10000]
+
+read await SEND_ASYNC_REQUEST
+
+read option zilla:flags "fin"
+read zilla:data.ext ${kafka:matchDataEx()
+                           .typeId(zilla:id("kafka"))
+                           .merged()
+                             .produce()
+                               .partition(-1, -1)
+                               .header("zilla:correlation-id", "59410e57-3e0f-4b61-9328-f645a7968ac8-55e623a5b35ce648e9af4ff07f8d307b")
+                               .build()
+                           .build()}
+read zilla:data.empty
+
+read closed
+write close
+
+
+accepted
+
+read zilla:begin.ext ${kafka:matchBeginEx()
+                              .typeId(zilla:id("kafka"))
+                              .merged()
+                                  .capabilities("FETCH_ONLY")
+                                  .topic("items-replies")
+                                  .partition(-1, -2)
+                                  .filter()
+                                    .header("zilla:correlation-id", "59410e57-3e0f-4b61-9328-f645a7968ac8-55e623a5b35ce648e9af4ff07f8d307b")
+                                    .build()
+                                  .build()
+                              .build()}
+
+connected
+
+write advise zilla:flush
+
+write notify SEND_ASYNC_REQUEST
+
+write zilla:data.ext ${kafka:dataEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                               .fetch()
+                                .partition(0, 1, 2)
+                                .progress(0, 2)
+                                .progress(1, 1)
+                                .key("92d0bf92-63e0-4cfc-ae73-71dee92d1544")
+                                .header(":status", "204")
+                                .header("zilla:correlation-id", "59410e57-3e0f-4b61-9328-f645a7968ac8-55e623a5b35ce648e9af4ff07f8d307b")
+                                .build()
+                            .build()}
+write flush
+
+write close
+read closed
+
+
+

specs/binding-http-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/http/kafka/streams/HttpIT.java

@@ -477,6 +477,15 @@ public void shouldPostItemCommand() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${http}/post.item.command.10k/client",
+        "${http}/post.item.command.10k/server"})
+    public void shouldPostItemCommand10k() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${http}/post.item.command.if.match/client",

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

@@ -432,6 +432,15 @@ public void shouldPostItemCommand() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${kafka}/post.item.command.10k/client",
+        "${kafka}/post.item.command.10k/server"})
+    public void shouldPostItemCommand10k() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${kafka}/post.item.async.command/client",