fix(binding-mqtt, binding-mqtt-kafka): prevent engine crash on MQTT CONNECT with Will message

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

@@ -3019,13 +3019,15 @@ protected final void doKafkaData(
             int limit,
             Flyweight extension)
         {
+            if (kafka != null)
+            {
+                doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                    traceId, authorization, budgetId, flags, reserved, buffer, offset, limit, extension);
 
-            doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
-                traceId, authorization, budgetId, flags, reserved, buffer, offset, limit, extension);
-
-            initialSeq += reserved;
+                initialSeq += reserved;
 
-            assert initialSeq - padding <= initialAck + initialMax;
+                assert initialSeq - padding <= initialAck + initialMax;
+            }
         }
 
         protected final void doKafkaData(
@@ -3037,12 +3039,15 @@ protected final void doKafkaData(
             OctetsFW payload,
             Flyweight extension)
         {
-            doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
-                traceId, authorization, budgetId, flags, reserved, payload, extension);
+            if (kafka != null)
+            {
+                doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                    traceId, authorization, budgetId, flags, reserved, payload, extension);
 
-            initialSeq += reserved;
+                initialSeq += reserved;
 
-            assert initialSeq <= initialAck + initialMax;
+                assert initialSeq <= initialAck + initialMax;
+            }
         }
 
         protected void doKafkaData(
@@ -3055,17 +3060,20 @@ protected void doKafkaData(
             Flyweight payload,
             Flyweight extension)
         {
-            final DirectBufferEx buffer = payload.buffer();
-            final int offset = payload.offset();
-            final int limit = payload.limit();
-            final int length = limit - offset;
+            if (kafka != null)
+            {
+                final DirectBufferEx buffer = payload.buffer();
+                final int offset = payload.offset();
+                final int limit = payload.limit();
+                final int length = limit - offset;
 
-            doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
-                traceId, authorization, budgetId, flags, reserved, buffer, offset, length, extension);
+                doData(kafka, originId, routedId, initialId, initialSeq, initialAck, initialMax,
+                    traceId, authorization, budgetId, flags, reserved, buffer, offset, length, extension);
 
-            initialSeq += reserved;
+                initialSeq += reserved;
 
-            assert initialSeq - padding <= initialAck + initialMax;
+                assert initialSeq - padding <= initialAck + initialMax;
+            }
         }
 
         private void doKafkaFlush(

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

@@ -275,6 +275,17 @@ public void shouldNotSendWillMessageOnWillIdMismatch() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("proxy.yaml")
+    @Configure(name = WILL_AVAILABLE_NAME, value = "false")
+    @Specification({
+        "${mqtt}/session.will.message.qos2.abort.before.session.established/client",
+        "${kafka}/session.will.message.qos2.abort.before.session.established/server"})
+    public void shouldSkipWillSignalOnQos2AbortBeforeSessionEstablished() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("proxy.yaml")
     @Configure(name = PUBLISH_MAX_QOS_NAME, value = "1")

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

@@ -3561,6 +3561,7 @@ private int onDecodeConnectWillPayload(
             final int willQos = decodeWillQos(connectFlags);
             final boolean willFlagSet = isSetWillFlag(connectFlags);
 
+            decode:
             if (willFlagSet && MqttState.initialOpened(session.state))
             {
                 int publishedWillSize = 0;
@@ -3588,6 +3589,12 @@ private int onDecodeConnectWillPayload(
 
                     final MqttWillMessageFW will = willMessageBuilder.build();
                     final int headerSize = willMessageBuilder.sizeof();
+
+                    if (!session.hasSessionWindow(headerSize))
+                    {
+                        break decode;
+                    }
+
                     int payloadSize = Math.min(limit - offset, session.initialBudget() - headerSize);
 
                     final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + payloadSize);

runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java

@@ -187,6 +187,16 @@ public void shouldSendWillMessage10k() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("server.yaml")
+    @Specification({
+        "${net}/session.will.message.header.exceeds.window/client",
+        "${app}/session.will.message.header.exceeds.window/server"})
+    public void shouldDeferWillMessageWhenHeaderExceedsSessionWindow() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("server.yaml")
     @Specification({

specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/client.rpt

@@ -0,0 +1,49 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+connect
+        "zilla://streams/kafka0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-messages")
+                                   .build()
+                               .build()}
+
+connected
+
+write await SESSION_META_STREAMS_OPENED
+write abort
+
+
+connect
+        "zilla://streams/kafka0"
+         option zilla:window 8192
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-retained")
+                                   .build()
+                               .build()}
+
+connected
+
+write await SESSION_META_STREAMS_OPENED
+write abort

specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/kafka/session.will.message.qos2.abort.before.session.established/server.rpt

@@ -0,0 +1,51 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+accept "zilla://streams/kafka0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-messages")
+                                   .build()
+                               .build()}
+
+connected
+
+read aborted
+
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                               .typeId(zilla:id("kafka"))
+                               .meta()
+                                   .topic("mqtt-retained")
+                                   .build()
+                               .build()}
+
+connected
+
+# partition metadata is still outstanding, no session state stream exists yet
+write notify SESSION_META_STREAMS_OPENED
+
+# no will signal and no expiry signal are produced,
+# the session was never established
+read aborted

specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/client.rpt

@@ -0,0 +1,38 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+# session stays half-duplex, the session reply stream is only opened once the
+# qos2 session state stream has been established
+connect "zilla://streams/mqtt0"
+         option zilla:window 8192
+         option zilla:transmission "half-duplex"
+
+write zilla:begin.ext ${mqtt:beginEx()
+                           .typeId(zilla:id("mqtt"))
+                           .session()
+                              .flags("WILL", "CLEAN_START")
+                              .expiry(1)
+                              .publishQosMax(2)
+                              .capabilities("REDIRECT")
+                              .clientId("client-1")
+                              .build()
+                           .build()}
+
+connected
+
+# abort while the qos2 partition metadata is still being negotiated,
+# no session state stream has been established yet
+write await SESSION_META_STREAMS_OPENED
+write abort

specs/binding-mqtt-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/mqtt/session.will.message.qos2.abort.before.session.established/server.rpt

@@ -0,0 +1,35 @@
+#
+# Copyright 2021-2026 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+accept "zilla://streams/mqtt0"
+        option zilla:window 8192
+        option zilla:transmission "half-duplex"
+
+accepted
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                             .typeId(zilla:id("mqtt"))
+                             .session()
+                                .flags("WILL", "CLEAN_START")
+                                .expiry(1)
+                                .publishQosMax(2)
+                                .capabilities("REDIRECT")
+                                .clientId("client-1")
+                                .build()
+                             .build()}
+
+connected
+
+read aborted

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

@@ -747,6 +747,15 @@ public void shouldGenerateLifeTimeIdOnCleanStart() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${kafka}/session.will.message.qos2.abort.before.session.established/client",
+        "${kafka}/session.will.message.qos2.abort.before.session.established/server"})
+    public void shouldSkipWillSignalOnQos2AbortBeforeSessionEstablished() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${kafka}/session.will.message.will.id.mismatch.skip.delivery/client",

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

@@ -630,6 +630,17 @@ public void shouldSaveWillMessage() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${mqtt}/session.will.message.qos2.abort.before.session.established/client",
+        "${mqtt}/session.will.message.qos2.abort.before.session.established/server"})
+    public void shouldSkipWillSignalOnQos2AbortBeforeSessionEstablished() throws Exception
+    {
+        k3po.start();
+        k3po.notifyBarrier("SESSION_META_STREAMS_OPENED");
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${mqtt}/publish.qos1/client",

specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/client.rpt

@@ -0,0 +1,45 @@
+#
+# 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 32
+         option zilla:transmission "duplex"
+
+write zilla:begin.ext ${mqtt:beginEx()
+                           .typeId(zilla:id("mqtt"))
+                           .session()
+                              .flags("WILL", "CLEAN_START")
+                              .clientId("one")
+                              .build()
+                           .build()}
+
+read zilla:begin.ext ${mqtt:matchBeginEx()
+                              .typeId(zilla:id("mqtt"))
+                              .session()
+                                .flags("WILL", "CLEAN_START")
+                                .subscribeQosMax(2)
+                                .publishQosMax(2)
+                                .packetSizeMax(33792)
+                                .capabilities("RETAIN", "WILDCARD", "SUBSCRIPTION_IDS", "SHARED_SUBSCRIPTIONS")
+                                .clientId("one")
+                                .build()
+                              .build()}
+
+connected
+
+read zilla:data.empty
+
+write abort