Connection cleanup can trigger exception

[akrambek]

Describe the bug
Debitor on stream close not being released and removing streams not related to the connection.

2023-09-24 10:07:07 started
2023-09-24 10:08:00 stopped
2023-09-24 10:08:00 org.agrona.concurrent.AgentTerminationException: java.lang.NullPointerException: Cannot invoke "java.util.function.LongConsumer.accept(long)" because "flusher" is null
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.doWork(DispatchAgent.java:681)
2023-09-24 10:08:00     at org.agrona.core/org.agrona.concurrent.AgentRunner.doDutyCycle(AgentRunner.java:291)
2023-09-24 10:08:00     at org.agrona.core/org.agrona.concurrent.AgentRunner.run(AgentRunner.java:164)
2023-09-24 10:08:00     at java.base/java.lang.Thread.run(Thread.java:1623)
2023-09-24 10:08:00 Caused by: java.lang.NullPointerException: Cannot invoke "java.util.function.LongConsumer.accept(long)" because "flusher" is null
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.budget.KafkaMergedBudget.flush(KafkaMergedBudget.java:182)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.budget.KafkaMergedBudget.credit(KafkaMergedBudget.java:66)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.budget.KafkaMergedBudgetCreditor.credit(KafkaMergedBudgetCreditor.java:67)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.onConnectionWindow(KafkaClientConnectionPool.java:1280)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.onConnectionMessage(KafkaClientConnectionPool.java:1110)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.handleReadInitial(DispatchAgent.java:1070)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.handleRead(DispatchAgent.java:1005)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.concurent.ManyToOneRingBuffer.read(ManyToOneRingBuffer.java:181)
2023-09-24 10:08:00     at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.doWork(DispatchAgent.java:675)
2023-09-24 10:08:00     ... 3 more
2023-09-24 10:08:00     Suppressed: java.lang.Exception: [engine/data#3]        [0x0303000000000181] streams=[consumeAt=0x00162568 (0x0000000000162568), produceAt=0x00162568 (0x0000000000162568)]
2023-09-24 10:08:00             at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.DispatchAgent.doWork(DispatchAgent.java:679)
2023-09-24 10:08:00             ... 3 more