fix(binding-kafka): emit a window when a produce flush is acknowledged

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

@@ -1456,10 +1456,8 @@ private void onClientInitialFlush(
                 }
             }
 
-            initialAck += reserved;
-
-            final int noAck = (int) (initialSeq - initialAck);
-            doClientInitialWindow(traceId, noAck, initialBudgetMax);
+            final int noAck = (int) (initialSeq - initialAck - reserved);
+            doClientInitialWindow(traceId, noAck, Math.max(initialMax, initialBudgetMax));
         }
 
         private void onClientInitialEnd(

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

@@ -249,6 +249,17 @@ public void shouldSendMessageValue10k() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("cache.yaml")
+    @Specification({
+        "${app}/message.value.after.flushes/client",
+        "${app}/message.value.after.flushes/server"})
+    @ScriptProperty("serverAddress \"zilla://streams/app1\"")
+    public void shouldSendMessageValueAfterFlushes() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("cache.yaml")
     @Specification({

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/DefaultZillaChannelConfig.java

@@ -22,6 +22,7 @@
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_BUDGET_ID;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_BYTE_ORDER;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_CAPABILITIES;
+import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_PADDABLE;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_PADDING;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_REDIRECTED_ID;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_SHARED_WINDOW;
@@ -35,7 +36,9 @@
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.util.Conversions.convertToInt;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.util.Conversions.convertToLong;
 
+import java.util.EnumSet;
 import java.util.Objects;
+import java.util.Set;
 
 import io.aklivity.k3po.runtime.driver.internal.netty.bootstrap.channel.DefaultChannelConfig;
 
@@ -47,6 +50,7 @@ public class DefaultZillaChannelConfig extends DefaultChannelConfig implements Z
     private long budgetId;
     private long streamId;
     private int padding;
+    private Set<ZillaFrameKind> paddable = EnumSet.of(ZillaFrameKind.DATA);
     private ZillaThrottleMode throttle = ZillaThrottleMode.STREAM;
     private ZillaUpdateMode update = ZillaUpdateMode.STREAM;
     private long affinity;
@@ -136,6 +140,23 @@ public int getPadding()
         return padding;
     }
 
+    @Override
+    public void setPaddable(String paddable)
+    {
+        final Set<ZillaFrameKind> kinds = EnumSet.noneOf(ZillaFrameKind.class);
+        for (String kind : paddable.trim().split("\\s+"))
+        {
+            kinds.add(ZillaFrameKind.decode(kind));
+        }
+        this.paddable = kinds;
+    }
+
+    @Override
+    public boolean isPaddable(ZillaFrameKind kind)
+    {
+        return paddable.contains(kind);
+    }
+
     @Override
     public void setUpdate(
         ZillaUpdateMode update)
@@ -253,6 +274,10 @@ else if (OPTION_PADDING.getName().equals(key))
         {
             setPadding(convertToInt(value));
         }
+        else if (OPTION_PADDABLE.getName().equals(key))
+        {
+            setPaddable(Objects.toString(value, "data"));
+        }
         else if (OPTION_UPDATE.getName().equals(key))
         {
             setUpdate(ZillaUpdateMode.decode(Objects.toString(value, null)));

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/DefaultZillaServerChannelConfig.java

@@ -21,6 +21,7 @@
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_BUDGET_ID;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_BYTE_ORDER;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_CAPABILITIES;
+import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_PADDABLE;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_PADDING;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_SHARED_WINDOW;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.types.ZillaTypeSystem.OPTION_STREAM_ID;
@@ -33,7 +34,9 @@
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.util.Conversions.convertToInt;
 import static io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.util.Conversions.convertToLong;
 
+import java.util.EnumSet;
 import java.util.Objects;
+import java.util.Set;
 
 import io.aklivity.k3po.runtime.driver.internal.netty.bootstrap.channel.DefaultServerChannelConfig;
 
@@ -45,6 +48,7 @@ public class DefaultZillaServerChannelConfig extends DefaultServerChannelConfig
     private long budgetId;
     private long streamId;
     private int padding;
+    private Set<ZillaFrameKind> paddable = EnumSet.of(ZillaFrameKind.DATA);
     private ZillaThrottleMode throttle = ZillaThrottleMode.STREAM;
     private ZillaUpdateMode update = ZillaUpdateMode.STREAM;
     private long affinity;
@@ -122,6 +126,23 @@ public int getPadding()
         return padding;
     }
 
+    @Override
+    public void setPaddable(String paddable)
+    {
+        final Set<ZillaFrameKind> kinds = EnumSet.noneOf(ZillaFrameKind.class);
+        for (String kind : paddable.trim().split("\\s+"))
+        {
+            kinds.add(ZillaFrameKind.decode(kind));
+        }
+        this.paddable = kinds;
+    }
+
+    @Override
+    public boolean isPaddable(ZillaFrameKind kind)
+    {
+        return paddable.contains(kind);
+    }
+
     @Override
     public void setUpdate(
         ZillaUpdateMode update)
@@ -253,6 +274,10 @@ else if (OPTION_PADDING.getName().equals(key))
         {
             setPadding(convertToInt(value));
         }
+        else if (OPTION_PADDABLE.getName().equals(key))
+        {
+            setPaddable(Objects.toString(value, "data"));
+        }
         else if (OPTION_UPDATE.getName().equals(key))
         {
             setUpdate(ZillaUpdateMode.decode(Objects.toString(value, null)));

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/ZillaChannelConfig.java

@@ -41,6 +41,10 @@ public interface ZillaChannelConfig extends ChannelConfig
 
     int getPadding();
 
+    void setPaddable(String paddable);
+
+    boolean isPaddable(ZillaFrameKind kind);
+
     void setUpdate(ZillaUpdateMode update);
 
     ZillaUpdateMode getUpdate();

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/ZillaFrameKind.java

@@ -0,0 +1,40 @@
+/*
+ * 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.
+ */
+package io.aklivity.zilla.runtime.engine.test.internal.k3po.ext.behavior;
+
+import java.util.Objects;
+
+public enum ZillaFrameKind
+{
+    DATA,
+    FLUSH;
+
+    public static ZillaFrameKind decode(
+        String value)
+    {
+        Objects.requireNonNull(value);
+
+        switch (value)
+        {
+        case "data":
+            return DATA;
+        case "flush":
+            return FLUSH;
+        default:
+            throw new IllegalArgumentException(value);
+        }
+    }
+}

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/behavior/ZillaTarget.java

@@ -489,6 +489,7 @@ private void doAdviseOutputFlush(
         final long authorization = channel.targetAuth();
         final int maximum = channel.targetMax();
         final long budgetId = channel.debitorId();
+        final int reservedBytes = channel.getConfig().isPaddable(ZillaFrameKind.FLUSH) ? channel.reservedBytes(0) : 0;
 
         final ChannelBuffer writeExt = channel.writeExtBuffer(FLUSH, true);
         final int writableExtBytes = writeExt.readableBytes();
@@ -505,11 +506,14 @@ private void doAdviseOutputFlush(
                 .traceId(supplyTraceId.getAsLong())
                 .authorization(authorization)
                 .budgetId(budgetId)
+                .reserved(reservedBytes)
                 .extension(ex -> ex.set(writeExtCopy))
                 .build();
 
         streamsBuffer.write(flush.typeId(), flush.buffer(), flush.offset(), flush.sizeof());
 
+        channel.writtenBytes(0, reservedBytes);
+
         writeExt.skipBytes(writableExtBytes);
         writeExt.discardReadBytes();
 
@@ -710,7 +714,8 @@ private boolean flushData(
         final boolean flushing = writeBuf == NULL_BUFFER;
         final int writableWin = channel.writableBytes();
         final int reservableBytes = Math.min(writeBuf.readableBytes(), writeBuffer.capacity() >> 1);
-        final int reservedBytes = writableWin > 0 || writeBuf.capacity() == 0 ? channel.reservedBytes(reservableBytes) : 0;
+        final int reservedBytes = channel.getConfig().isPaddable(ZillaFrameKind.DATA) &&
+                (writableWin > 0 || writeBuf.capacity() == 0) ? channel.reservedBytes(reservableBytes) : 0;
         final int writableBytes = Math.max(Math.min(reservedBytes - channel.targetPad(), writeBuf.readableBytes()), 0);
 
         // allow extension-only DATA frames to be flushed immediately

runtime/engine/src/test/java/io/aklivity/zilla/runtime/engine/test/internal/k3po/ext/types/ZillaTypeSystem.java

@@ -37,6 +37,7 @@ public final class ZillaTypeSystem implements TypeSystemSpi
     public static final TypeInfo<Long> OPTION_BUDGET_ID = new TypeInfo<>("budgetId", Long.class);
     public static final TypeInfo<Long> OPTION_STREAM_ID = new TypeInfo<>("streamId", Long.class);
     public static final TypeInfo<Integer> OPTION_PADDING = new TypeInfo<>("padding", Integer.class);
+    public static final TypeInfo<String> OPTION_PADDABLE = new TypeInfo<>("paddable", String.class);
     public static final TypeInfo<String> OPTION_UPDATE = new TypeInfo<>("update", String.class);
     public static final TypeInfo<String> OPTION_TRANSMISSION = new TypeInfo<>("transmission", String.class);
     public static final TypeInfo<String> OPTION_THROTTLE = new TypeInfo<>("throttle", String.class);
@@ -90,6 +91,7 @@ public ZillaTypeSystem()
         acceptOptions.add(OPTION_SHARED_WINDOW);
         acceptOptions.add(OPTION_BUDGET_ID);
         acceptOptions.add(OPTION_PADDING);
+        acceptOptions.add(OPTION_PADDABLE);
         acceptOptions.add(OPTION_UPDATE);
         acceptOptions.add(OPTION_AUTHORIZATION);
         acceptOptions.add(OPTION_THROTTLE);
@@ -110,6 +112,7 @@ public ZillaTypeSystem()
         connectOptions.add(OPTION_BUDGET_ID);
         connectOptions.add(OPTION_STREAM_ID);
         connectOptions.add(OPTION_PADDING);
+        connectOptions.add(OPTION_PADDABLE);
         connectOptions.add(OPTION_UPDATE);
         connectOptions.add(OPTION_AUTHORIZATION);
         connectOptions.add(OPTION_THROTTLE);

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

@@ -0,0 +1,83 @@
+#
+# 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.
+#
+
+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:padding 4096
+    option zilla:paddable "data flush"
+    option zilla:transmission "half-duplex"
+    option zilla:affinity 177
+
+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 advise zilla:flush
+
+write advise zilla:flush
+
+write zilla:data.ext ${kafka:dataEx()
+                              .typeId(zilla:id("kafka"))
+                              .produce()
+                                  .timestamp(1716424650323)
+                                  .build()
+                              .build()}
+write "Hello, world"
+write flush