Running emqtt_bench triggers exception in connection pool

[akrambek]

To Reproduce
Steps to reproduce the behavior:

1. Checkout https://github.com/aklivity/zilla-examples/tree/main/mqtt.kafka.broker and start the stack
2. Download the correct emqx-bench binary: https://github.com/emqx/emqtt-bench/releases
3. ./bin/emqtt_bench pub -c 100 -I 10 -t bench/%i -s 256 --shortids

7183org.agrona.concurrent.AgentTerminationException: java.lang.NullPointerException: Cannot read field "initialAck" because "stream" is null
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:707)
	at org.agrona.core/org.agrona.concurrent.AgentRunner.doDutyCycle(AgentRunner.java:291)
	at org.agrona.core/org.agrona.concurrent.AgentRunner.run(AgentRunner.java:164)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: java.lang.NullPointerException: Cannot read field "initialAck" because "stream" is null
	at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.doStreamWindow(KafkaClientConnectionPool.java:1755)
	at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.onConnectionWindow(KafkaClientConnectionPool.java:1732)
	at io.aklivity.zilla.runtime.binding.kafka/io.aklivity.zilla.runtime.binding.kafka.internal.stream.KafkaClientConnectionPool$KafkaClientConnection.onConnectionMessage(KafkaClientConnectionPool.java:1504)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleReadInitial(EngineWorker.java:1106)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleRead(EngineWorker.java:1041)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.concurent.ManyToOneRingBuffer.read(ManyToOneRingBuffer.java:181)
	at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:701)
	... 3 more
	Suppressed: java.lang.Exception: [engine/data#9]	[0x090900000000059b] streams=[consumeAt=0x00090970 (0x0000000000490970), produceAt=0x00090d90 (0x0000000000490d90)]
		at io.aklivity.zilla.runtime.engine/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:705)
		... 3 more

[jfallows]

Closing this in favor of #762 which has same stack trace and reproduction recipe.