Fix virtual cluster produce under load

[bmaidics]

In our kafka-proxy virtual cluster example, if we produce with higher throughput, Kafka tears down the connection as it fails to parse.

Reproduction:
./kafka-producer-perf-test.sh --topic test2 --throughput 8224 --num-records 2500 --record-size 255 --producer-props bootstrap.servers=kafka.external.net:10000

Kafka exceptions:

2025-10-26 12:56:06 org.apache.kafka.common.errors.InvalidRequestException: Error parsing request header. Our best guess of the apiKey is: 3
2025-10-26 12:56:06 Caused by: java.lang.RuntimeException: Error reading byte array of 32 byte(s): only 21 byte(s) available
2025-10-26 12:56:06     at org.apache.kafka.common.protocol.ByteBufferAccessor.readArray(ByteBufferAccessor.java:60)
2025-10-26 12:56:06     at org.apache.kafka.common.protocol.Readable.readString(Readable.java:43)
2025-10-26 12:56:06     at org.apache.kafka.common.message.RequestHeaderData.read(RequestHeaderData.java:122)
2025-10-26 12:56:06     at org.apache.kafka.common.message.RequestHeaderData.<init>(RequestHeaderData.java:84)
2025-10-26 12:56:06     at org.apache.kafka.common.requests.RequestHeader.parse(RequestHeader.java:125)
2025-10-26 12:56:06     at kafka.network.Processor.parseRequestHeader(SocketServer.scala:1117)
2025-10-26 12:56:06     at kafka.network.Processor.$anonfun$processCompletedReceives$1(SocketServer.scala:1130)
2025-10-26 12:56:06     at java.base/java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:647)
2025-10-26 12:56:06     at kafka.network.Processor.processCompletedReceives(SocketServer.scala:1126)
2025-10-26 12:56:06     at kafka.network.Processor.run(SocketServer.scala:1012)
2025-10-26 12:56:06     at java.base/java.lang.Thread.run(Thread.java:840)

2025-10-26 13:06:58 Caused by: java.lang.IllegalArgumentException: newLimit > capacity: (16165 > 16103)
2025-10-26 13:06:58     at java.base/java.nio.Buffer.createLimitException(Buffer.java:395)
2025-10-26 13:06:58     at java.base/java.nio.Buffer.limit(Buffer.java:369)
2025-10-26 13:06:58     at java.base/java.nio.ByteBuffer.limit(ByteBuffer.java:1529)
2025-10-26 13:06:58     at java.base/java.nio.ByteBuffer.limit(ByteBuffer.java:267)
2025-10-26 13:06:58     at org.apache.kafka.common.protocol.ByteBufferAccessor.readByteBuffer(ByteBufferAccessor.java:76)
2025-10-26 13:06:58     at org.apache.kafka.common.protocol.Readable.readRecords(Readable.java:61)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$PartitionProduceData.read(ProduceRequestData.java:825)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$PartitionProduceData.<init>(ProduceRequestData.java:790)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$TopicProduceData.read(ProduceRequestData.java:534)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$TopicProduceData.<init>(ProduceRequestData.java:464)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData.read(ProduceRequestData.java:187)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData.<init>(ProduceRequestData.java:114)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.ProduceRequest.parse(ProduceRequest.java:257)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.AbstractRequest.doParseRequest(AbstractRequest.java:171)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.AbstractRequest.parseRequest(AbstractRequest.java:165)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.RequestContext.parseRequest(RequestContext.java:95)
2025-10-26 13:06:58     at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:102)
2025-10-26 13:06:58     at kafka.network.Processor.$anonfun$processCompletedReceives$1(SocketServer.scala:1148)
2025-10-26 13:06:58     at java.base/java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:647)
2025-10-26 13:06:58     at kafka.network.Processor.processCompletedReceives(SocketServer.scala:1126)
2025-10-26 13:06:58     at kafka.network.Processor.run(SocketServer.scala:1012)
2025-10-26 13:06:58     at java.base/java.lang.Thread.run(Thread.java:840)
2025-10-26 13:06:58 [2025-10-26 20:06:58,852] ERROR Exception while processing request from 172.18.0.2:9092-172.18.0.4:33550-259 (kafka.network.Processor)
2025-10-26 13:06:58 org.apache.kafka.common.errors.InvalidRequestException: Error getting request for apiKey: PRODUCE, apiVersion: 8, connectionId: 172.18.0.2:9092-172.18.0.4:33550-259, listenerName: ListenerName(PLAINTEXT), principal: User:ANONYMOUS
2025-10-26 13:06:58 Caused by: java.lang.IllegalArgumentException: newLimit > capacity: (16165 > 16103)
2025-10-26 13:06:58     at java.base/java.nio.Buffer.createLimitException(Buffer.java:395)
2025-10-26 13:06:58     at java.base/java.nio.Buffer.limit(Buffer.java:369)
2025-10-26 13:06:58     at java.base/java.nio.ByteBuffer.limit(ByteBuffer.java:1529)
2025-10-26 13:06:58     at java.base/java.nio.ByteBuffer.limit(ByteBuffer.java:267)
2025-10-26 13:06:58     at org.apache.kafka.common.protocol.ByteBufferAccessor.readByteBuffer(ByteBufferAccessor.java:76)
2025-10-26 13:06:58     at org.apache.kafka.common.protocol.Readable.readRecords(Readable.java:61)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$PartitionProduceData.read(ProduceRequestData.java:825)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$PartitionProduceData.<init>(ProduceRequestData.java:790)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$TopicProduceData.read(ProduceRequestData.java:534)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData$TopicProduceData.<init>(ProduceRequestData.java:464)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData.read(ProduceRequestData.java:187)
2025-10-26 13:06:58     at org.apache.kafka.common.message.ProduceRequestData.<init>(ProduceRequestData.java:114)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.ProduceRequest.parse(ProduceRequest.java:257)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.AbstractRequest.doParseRequest(AbstractRequest.java:171)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.AbstractRequest.parseRequest(AbstractRequest.java:165)
2025-10-26 13:06:58     at org.apache.kafka.common.requests.RequestContext.parseRequest(RequestContext.java:95)
2025-10-26 13:06:58     at kafka.network.RequestChannel$Request.<init>(RequestChannel.scala:102)
2025-10-26 13:06:58     at kafka.network.Processor.$anonfun$processCompletedReceives$1(SocketServer.scala:1148)
2025-10-26 13:06:58     at java.base/java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:647)
2025-10-26 13:06:58     at kafka.network.Processor.processCompletedReceives(SocketServer.scala:1126)
2025-10-26 13:06:58     at kafka.network.Processor.run(SocketServer.scala:1012)
2025-10-26 13:06:58     at java.base/java.lang.Thread.run(Thread.java:840)

[bmaidics]

Fixed by: https://github.com/aklivity/zilla-plus/pull/719