[Connection Pool] binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.doConnectionWindow(KafkaClientConnectionPool.java:1318)

[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 12:49:29 Caused by: java.util.NoSuchElementException
2023-11-02 12:49:29     at org.agrona.core/org.agrona.collections.LongArrayQueue.removeLong(LongArrayQueue.java:319)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.doConnectionWindow(KafkaClientConnectionPool.java:1318)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientStream.flushStreamWindow(KafkaClientConnectionPool.java:890)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientStream.doStreamData(KafkaClientConnectionPool.java:865)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.onConnectionData(KafkaClientConnectionPool.java:1441)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.onConnectionMessage(KafkaClientConnectionPool.java:1364)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.handleReadReply(DispatchAgent.java:1244)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.handleRead(DispatchAgent.java:1045)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.concurent.ManyToOneRingBuffer.read(ManyToOneRingBuffer.java:181)
2023-11-02 12:49:29     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.doWork(DispatchAgent.java:701)
2023-11-02 12:49:29     ... 3 more
2023-11-02 12:49:29     Suppressed: java.lang.Exception: [engine/data#9]        [0x090900000000013a] streams=[consumeAt=0x0034a8e0 (0x0000000024b4a8e0), produceAt=0x0034a8e0 (0x0000000024b4a8e0)]
2023-11-02 12:49:29             at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.doWork(DispatchAgent.java:705)
2023-11-02 12:49:29             ... 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