[MQTT-Kafka] Exception runtime.binding.mqtt.kafka.internal.types.MqttExpirySignalFW.wrap(MqttExpirySignalFW.java:45)

[akrambek]

Describe the bug
Running the load-test branch of zilla-demos/taxi I would periodically run into the below exception. Some happen on startup, and some happen when the taxi-service (the MQTT producer) is stopped or restarted.

2023-11-02 11:34:56 Caused by: java.lang.IndexOutOfBoundsException: limit=13404 is beyond maxLimit=103
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.Flyweight.checkLimit(Flyweight.java:67)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.String16FW.wrap(String16FW.java:92)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.MqttExpirySignalFW.wrap(MqttExpirySignalFW.java:45)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.types.MqttSessionSignalFW.wrap(MqttSessionSignalFW.java:81)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSessionFactory$KafkaSignalStream.onKafkaData(MqttKafkaSessionFactory.java:1188)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.binding.mqtt.kafka/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSessionFactory$KafkaSignalStream.onSignalMessage(MqttKafkaSessionFactory.java:1013)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.handleReadReply(DispatchAgent.java:1244)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.handleRead(DispatchAgent.java:1045)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.concurent.ManyToOneRingBuffer.read(ManyToOneRingBuffer.java:181)
2023-11-02 11:34:56     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.doWork(DispatchAgent.java:701)
2023-11-02 11:34:56     ... 3 more
2023-11-02 11:34:56     Suppressed: java.lang.Exception: [engine/data#0]        [0x0000000000000002] streams=[consumeAt=0x003ceb38 (0x0000000000fceb38), produceAt=0x003f5de0 (0x0000000000ff5de0)]
2023-11-02 11:34:56             at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.doWork(DispatchAgent.java:705)
2023-11-02 11:34:56             ... 3 more

To Reproduce
Steps to reproduce the behavior:

1. checkout zilla-demos/taxi@mqtt-load-test
2. start the demo ./startup.sh
3. Once the setup is running stable the metric stream_active_received{namespace="zilla-taxi-demo",binding="mqtt_server"} should be 500
4. Let it run

Desktop (please complete the following information):

• OS: MacOS
• Zilla: 0.9.55