fix(binding-kafka): guard budget credit against released creditor slot

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java

@@ -779,7 +779,10 @@ private void onClientInitialData(
                 onClientFanMemberClosed(traceId, stream);
             }
 
-            creditor.credit(traceId, partitionIndex, reserved);
+            if (partitionIndex != NO_CREDITOR_INDEX)
+            {
+                creditor.credit(traceId, partitionIndex, reserved);
+            }
         }
 
         private void onClientInitialFlush(
@@ -823,7 +826,11 @@ stream.entryMark, stream.valueMark, stream.valueLimit, now().toEpochMilli(), str
                 stream.cleanupClient(traceId, error);
                 onClientFanMemberClosed(traceId, stream);
             }
-            creditor.credit(traceId, partitionIndex, reserved);
+
+            if (partitionIndex != NO_CREDITOR_INDEX)
+            {
+                creditor.credit(traceId, partitionIndex, reserved);
+            }
         }
 
         private void flushClientFanInitialIfNecessary(
@@ -1404,7 +1411,7 @@ private void onClientInitialFlush(
                     doClientReplyAbortIfNecessary(traceId);
                     fan.onClientFanMemberClosed(traceId, this);
                 }
-                else
+                else if (!KafkaState.initialClosed(state))
                 {
                     fan.onClientInitialFlush(this, flush);
                 }

runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheProduceIT.java

@@ -225,6 +225,17 @@ public void shouldSendMessageValueNull() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("cache.yaml")
+    @Specification({
+        "${app}/message.value.after.fan.reply.abort/client",
+        "${app}/message.value.after.fan.reply.abort/server"})
+    @ScriptProperty("serverAddress \"zilla://streams/app1\"")
+    public void shouldSendMessageValueAfterFanReplyAbort() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("cache.yaml")
     @Specification({

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.value.after.fan.reply.abort/client.rpt

@@ -0,0 +1,90 @@
+#
+# Copyright 2021-2026 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+property deltaMillis 0L
+
+connect "zilla://streams/app0"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("test")
+                                   .build()
+                               .build()}
+
+connected
+
+read zilla:begin.ext ${kafka:beginEx()
+                              .typeId(zilla:id("kafka"))
+                              .meta()
+                                  .topic("test")
+                                  .build()
+                              .build()}
+
+read zilla:data.ext ${kafka:dataEx()
+                             .typeId(zilla:id("kafka"))
+                             .meta()
+                                 .partition(0, 177)
+                                 .build()
+                             .build()}
+
+read notify ROUTED_BROKER_CLIENT
+
+connect await ROUTED_BROKER_CLIENT
+        "zilla://streams/app0"
+    option zilla:window 8192
+    option zilla:transmission "half-duplex"
+    option zilla:affinity 0xb1
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .produce()
+                                   .topic("test")
+                                   .partition(0)
+                                   .build()
+                               .build()}
+
+connected
+
+read zilla:begin.ext ${kafka:beginEx()
+                              .typeId(zilla:id("kafka"))
+                              .produce()
+                                  .topic("test")
+                                  .partition(0)
+                                  .build()
+                              .build()}
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .produce()
+                                  .timestamp(1716424650323)
+                                  .build()
+                              .build()}
+write "Hello, world"
+write flush
+
+read aborted
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .produce()
+                                  .timestamp(1716424650323)
+                                  .build()
+                              .build()}
+write "Hello, again"
+write flush

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.value.after.fan.reply.abort/server.rpt

@@ -0,0 +1,77 @@
+#
+# Copyright 2021-2026 Aklivity Inc.
+#
+# Aklivity licenses this file to you under the Apache License,
+# version 2.0 (the "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at:
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+
+property serverAddress "zilla://streams/app0"
+
+accept ${serverAddress}
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+    option zilla:update "handshake"
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                              .typeId(zilla:id("kafka"))
+                              .meta()
+                                  .topic("test")
+                                  .build()
+                              .build()}
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("test")
+                                   .build()
+                               .build()}
+
+connected
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .meta()
+                                  .partition(0, 177)
+                                  .build()
+                              .build()}
+write flush
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                              .typeId(zilla:id("kafka"))
+                              .produce()
+                                  .topic("test")
+                                  .partition(0)
+                                  .build()
+                              .build()}
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .produce()
+                                   .topic("test")
+                                   .partition(0)
+                                   .build()
+                               .build()}
+
+connected
+
+read zilla:data.ext ${kafka:matchDataEx()
+                             .typeId(zilla:id("kafka"))
+                             .produce()
+                                 .build()
+                             .build()}
+read "Hello, world"
+
+write abort

specs/binding-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/kafka/streams/application/ProduceIT.java

@@ -47,6 +47,15 @@ public void shouldReconnectIdleNoErrorParallel() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${app}/message.value.after.fan.reply.abort/client",
+        "${app}/message.value.after.fan.reply.abort/server"})
+    public void shouldSendMessageValueAfterFanReplyAbort() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${app}/topic.missing/client",