fix(binding-mqtt-kafka): account for the retained stream when granting the publish window

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

@@ -828,9 +828,18 @@ private void doMqttWindow(
             int padding,
             int capabilities)
         {
-            final boolean retainedFlag = hasPublishFlagRetained(publishFlags);
-            final long newInitialAck = retainedFlag ? Math.min(messages.initialAck, retained.initialAck) : messages.initialAck;
-            final int newInitialMax = retainedFlag ? Math.max(messages.initialMax, retained.initialMax) : messages.initialMax;
+            // messages and retained advertise credit differently - messages holds its ack at zero
+            // and grows its max, retained advances its ack against a fixed max - so minimizing ack
+            // and max independently would combine the ack of one with the max of the other and cap
+            // the stream for its whole lifetime; compare the budgets each one actually offers
+            final int messagesBudget = messages.initialMax - (int)(messages.initialSeq - messages.initialAck);
+            final int retainedBudget = retained.initialMax - (int)(retained.initialSeq - retained.initialAck);
+            final int newInitialBudget = retainAvailable ?
+                Math.min(messagesBudget, retainedBudget) : messagesBudget;
+            final int newInitialMax = retainAvailable ?
+                Math.min(messages.initialMax, retained.initialMax) : messages.initialMax;
+            final long newInitialAck =
+                Math.max(initialAck, initialSeq - Math.max(newInitialMax - newInitialBudget, 0));
 
             if (MqttKafkaState.initialOpened(messages.state) &&
                 (!retainAvailable || MqttKafkaState.initialOpened(retained.state)) &&

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

@@ -246,6 +246,17 @@ public void shouldPublishRetainedMessage() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("proxy.yaml")
+    @Configure(name = WILL_AVAILABLE_NAME, value = "false")
+    @Specification({
+        "${mqtt}/publish.many.messages.retain.available/client",
+        "${kafka}/publish.many.messages.retain.available/server"})
+    public void shouldPublishManyMessagesRetainAvailable() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("proxy.yaml")
     @Configure(name = WILL_AVAILABLE_NAME, value = "false")

specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java

@@ -171,6 +171,15 @@ public void shouldPublishRetainedMessage() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${kafka}/publish.many.messages.retain.available/client",
+        "${kafka}/publish.many.messages.retain.available/server"})
+    public void shouldPublishManyMessagesRetainAvailable() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${kafka}/publish.multiple.messages/client",

specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java

@@ -180,6 +180,15 @@ public void shouldPublishRetainedMessage() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${mqtt}/publish.many.messages.retain.available/client",
+        "${mqtt}/publish.many.messages.retain.available/server"})
+    public void shouldPublishManyMessagesRetainAvailable() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${mqtt}/publish.with.user.properties.distinct/client",