Running emqtt_bench sub triggers an exception

[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 ./emqtt_bench pub -c 100 -I 10 -t bench/%i -s 256 --shortids -p 7183
4. Run ./emqtt_bench sub -c 100 -t bench/%i -s 256 --shortids -p 7183
5. See error

org.agrona.concurrent.AgentTerminationException: java.lang.NegativeArraySizeException: -508
	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.NegativeArraySizeException: -508
	at org.agrona.core/org.agrona.BitUtil.toHexByteArray(BitUtil.java:218)
	at org.agrona.core/org.agrona.BitUtil.toHex(BitUtil.java:283)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSubscribeFactory.offsetMetadataListToString(MqttKafkaSubscribeFactory.java:1849)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSubscribeFactory$KafkaMessagesProxy.lambda$doKafkaConsumerFlush$17(MqttKafkaSubscribeFactory.java:1609)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.stream.KafkaMergedConsumerFlushExFW$Builder.progress(KafkaMergedConsumerFlushExFW.java:80)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSubscribeFactory$KafkaMessagesProxy.lambda$doKafkaConsumerFlush$18(MqttKafkaSubscribeFactory.java:1604)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.stream.KafkaMergedFlushExFW$Builder.consumer(KafkaMergedFlushExFW.java:180)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSubscribeFactory$KafkaMessagesProxy.lambda$doKafkaConsumerFlush$19(MqttKafkaSubscribeFactory.java:1602)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.stream.KafkaFlushExFW$Builder.merged(KafkaFlushExFW.java:317)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSubscribeFactory$KafkaMessagesProxy.doKafkaConsumerFlush(MqttKafkaSubscribeFactory.java:1602)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSubscribeFactory$MqttSubscribeProxy.commitOffset(MqttKafkaSubscribeFactory.java:815)
	at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSubscribeFactory$MqttSubscribeProxy.onMqttFlush(MqttKafkaSubscribeFactory.java:536)

Expected behavior
No exception

https://drive.google.com/open?id=1wk4g0O0JjsrlUOgOHn_Rr6mCH6FBUQK3&usp=drive_copy

[jfallows]

Potentially related to #999

[bmaidics]

Not related to the mentioned ticket, and only reproducible using qos2 produce and subscribe:
Run ./emqtt_bench pub -c 100 -q 2 -I 10 -t bench/%i -s 256 --shortids -p 7183
Run ./emqtt_bench sub -c 100 -q 2 -t bench/%i -s 256 --shortids -p 7183