MQTT client connections cause errors/crashes

[vordimous]

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

The doStreamSignalNow(long, int)" because "stream" is null error happened pretty consistently after the demo was running stable for about 5 min.

java.util.NoSuchElementException
java.lang.IndexOutOfBoundsException: limit=860 is beyond maxLimit=512
java.lang.IndexOutOfBoundsException: limit=717 is beyond maxLimit=512
java.lang.IndexOutOfBoundsException: limit=700 is beyond maxLimit=512
java.lang.IndexOutOfBoundsException: limit=562 is beyond maxLimit=512
java.lang.IllegalArgumentException: invalid offset: -65536
java.lang.NullPointerException: Cannot invoke "io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientStream.doStreamSignalNow(long, int)" because "stream" is null

To Reproduce
Steps to reproduce the behavior:

1. checkout zilla-demos/taxi@mqtt-load-test
2. start the demo ./startup.sh
3. Some errors will happen as soon as everything starts up. Zilla will automatically restart.
4. Stop and restart the taxi-service: docker-compose -p zilla-taxi-demo restart taxi-service
5. Once the setup is running stable the metric stream_active_received{namespace="zilla-taxi-demo",binding="mqtt_server"} should be 300
6. After a few min the "stream" is null error pops up.

Desktop (please complete the following information):

• OS: MacOS
• Zilla: 0.9.55