Running emqtt_bench triggers an exception in mqtt during the decoding

[akrambek]

Describe the bug
A clear and concise description of what the bug is.

To Reproduce
Steps to reproduce the behavior:

1. Start https://github.com/aklivity/zilla-examples/tree/main/mqtt.kafka.broker
2. Download the emqtt_bench tool
3. Run test ./emqtt_bench pub -c 100 -I 10 -t bench/%i -s 256 --shortids -p 7183
4. See error

org.agrona.concurrent.AgentTerminationException: java.lang.IndexOutOfBoundsException: offset=17362459 is beyond maxLimit=17359387
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:823)
	at org.agrona.core/org.agrona.concurrent.AgentRunner.doWork(AgentRunner.java:304)
	at org.agrona.core/org.agrona.concurrent.AgentRunner.workLoop(AgentRunner.java:296)
	at org.agrona.core/org.agrona.concurrent.AgentRunner.run(AgentRunner.java:162)
	at java.base/java.lang.Thread.run(Thread.java:1589)
Caused by: java.lang.IndexOutOfBoundsException: offset=17362459 is beyond maxLimit=17359387
	at io.aklivity.zilla.runtime.binding.mqtt/io.aklivity.zilla.runtime.binding.mqtt.internal.types.Flyweight.wrap(Flyweight.java:51)
	at io.aklivity.zilla.runtime.binding.mqtt/io.aklivity.zilla.runtime.binding.mqtt.internal.types.OctetsFW.wrap(OctetsFW.java:39)
	at io.aklivity.zilla.runtime.binding.mqtt/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory.decodePublishPayload(MqttServerFactory.java:1484)
	at io.aklivity.zilla.runtime.binding.mqtt/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer.decodeNetwork(MqttServerFactory.java:4804)
	at io.aklivity.zilla.runtime.binding.mqtt/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer.onNetworkData(MqttServerFactory.java:2612)
	at io.aklivity.zilla.runtime.binding.mqtt/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer.onNetwork(MqttServerFactory.java:2527)
	at io.aklivity.zilla.runtime.binding.mqtt/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory.lambda$newStream$1(MqttServerFactory.java:569)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleReadInitial(EngineWorker.java:1238)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleRead(EngineWorker.java:1205)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.concurent.ManyToOneRingBuffer.read(ManyToOneRingBuffer.java:229)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:817)
	... 4 more
	Suppressed: java.lang.Exception: [engine/data#9]	[0x0909000000000859] streams=[consumeAt=0x010922d0 (0x00000003070922d0), produceAt=0x011d3078 (0x00000003071d3078)]
		at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:821)
		... 4 more
stopped

Expected behavior
No exception

[jfallows]

Unable to reproduce, need dump to further diagnose.
Please reopen if reproduced.