Zilla intermittently crashes with IndexOutOfBoundsException at high mqtt load

[bmaidics]

Describe the bug
Sometimes Zilla crashes with IndexOutOfBoundsException when performing MQTT load test.

To Reproduce
Steps to reproduce the behavior:

1. Run Zilla locally with basic mqtt-kafka broker config
2. Run emqtt bench pub until the the benchmark hangs (you might need to stop and run multiple times):
   ./emqtt_bench pub -h localhost -p 1883 -q 0 -c 100 -I 10 -L 50000 -t bench/%i
3. See error

org.agrona.concurrent.AgentTerminationException: java.lang.IndexOutOfBoundsException: index=-56 length=8 capacity=4194304
        at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:841)
        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:833)
Caused by: java.lang.IndexOutOfBoundsException: index=-56 length=8 capacity=4194304
        at org.agrona.core/org.agrona.AbstractMutableDirectBuffer.boundsCheck0(AbstractMutableDirectBuffer.java:1719)
        at org.agrona.core/org.agrona.concurrent.UnsafeBuffer.getAndAddLong(UnsafeBuffer.java:456)
        at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.budget.DefaultBudgetCreditor.credit(DefaultBudgetCreditor.java:139)
        at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaCacheClientProduceFactory$KafkaCacheClientProduceFan.onClientInitialData(KafkaCacheClientProduceFactory.java:774)
        at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaCacheClientProduceFactory$KafkaCacheClientProduceStream.onClientInitialData(KafkaCacheClientProduceFactory.java:1354)
        at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaCacheClientProduceFactory$KafkaCacheClientProduceStream.onClientMessage(KafkaCacheClientProduceFactory.java:1276)
        at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer.lambda$andThen$0(MessageConsumer.java:82)
        at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleReadInitial(EngineWorker.java:1260)
        at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleRead(EngineWorker.java:1227)
        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:835)
        ... 4 more
        Suppressed: java.lang.Exception: [engine/data#9]        [0x0909000000001755] streams=[consumeAt=0x0079b190 (0x0000000009f9b190), produceAt=0x007a0728 (0x0000000009fa0728)]
                at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:839)
                ... 4 more