From dd85ea908c1022eeaa49ea9ce39ca6868d31858d Mon Sep 17 00:00:00 2001 From: Sascha Freiheit Date: Tue, 25 Aug 2026 16:01:12 +0200 Subject: [PATCH 1/2] fix(binding-mqtt): guard will-message header write against insufficient session window When a client sends CONNECT with a Will message, onDecodeConnectWillPayload computed payloadSize as session.initialBudget() - headerSize with no floor. The guarding MqttState.initialOpened(session.state) check only proves a WINDOW frame arrived at least once for the session stream, not that the granted window is large enough to hold the encoded will-message header. When it isn't, the subtraction goes negative, producing a negative OctetsFW.wrap() limit and an uncaught IndexOutOfBoundsException that terminates the engine worker. Add a session.hasSessionWindow(headerSize) check before building the header, mirroring the same pattern already used by the SUBSCRIBE-to-session-state sync path. When the window is too small, return progress unchanged so the existing decodeSlot buffering parks the bytes and decodeNetwork retries automatically on the next session WINDOW. Fixes #2428 Co-Authored-By: Claude Sonnet 5 --- .../internal/stream/MqttServerFactory.java | 7 +++ .../internal/stream/server/v5/SessionIT.java | 10 ++++ .../client.rpt | 45 ++++++++++++++++ .../server.rpt | 52 +++++++++++++++++++ .../client.rpt | 41 +++++++++++++++ .../server.rpt | 42 +++++++++++++++ .../mqtt/streams/application/SessionIT.java | 9 ++++ .../mqtt/streams/network/v5/SessionIT.java | 9 ++++ 8 files changed, 215 insertions(+) create mode 100644 specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/client.rpt create mode 100644 specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/server.rpt create mode 100644 specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/client.rpt create mode 100644 specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/server.rpt diff --git a/runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java b/runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java index 4fd2e98ef6d..ec74d626047 100644 --- a/runtime/binding-mqtt/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/MqttServerFactory.java +++ b/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); diff --git a/runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java b/runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java index 0e822225360..7c82a3e1984 100644 --- a/runtime/binding-mqtt/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/internal/stream/server/v5/SessionIT.java +++ b/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({ diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/client.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/client.rpt new file mode 100644 index 00000000000..0d0462aa662 --- /dev/null +++ b/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 diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/server.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/server.rpt new file mode 100644 index 00000000000..a32f839eec1 --- /dev/null +++ b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/application/session.will.message.header.exceeds.window/server.rpt @@ -0,0 +1,52 @@ +# +# 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. +# + +# window is smaller than the encoded will message header (44 bytes for topic +# "wills/one"), so the will message must remain deferred instead of being +# written with a negative payload size + +accept "zilla://streams/app0" + option zilla:window 32 + option zilla:transmission "duplex" + +accepted + +read zilla:begin.ext ${mqtt:matchBeginEx() + .typeId(zilla:id("mqtt")) + .session() + .flags("WILL", "CLEAN_START") + .clientId("one") + .build() + .build()} + +write zilla:begin.ext ${mqtt:beginEx() + .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 + +write zilla:data.empty +write flush + +read aborted diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/client.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/client.rpt new file mode 100644 index 00000000000..090baa0ac9d --- /dev/null +++ b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/client.rpt @@ -0,0 +1,41 @@ +# +# 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/net0" + option zilla:window 8192 + option zilla:transmission "duplex" + option zilla:byteorder "network" + +connected + +write [0x10 0x2f] # CONNECT + [0x00 0x04] "MQTT" # protocol name + [0x05] # protocol version + [0x06] # flags = will flag, clean start + [0x00 0x0a] # keep alive = 10s + [0x05] # properties + [0x27] 33792 # maximum packet size = 33792 + [0x00 0x03] "one" # client id + [0x00] # will properties + [0x00 0x09] "wills/one" # will topic + [0x00 0x0c] "will payload" # will payload + +read [0x20 0x03] # CONNACK + [0x00] # flags = none + [0x00] # reason code + [0x00] # properties = none + +write abort diff --git a/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/server.rpt b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/server.rpt new file mode 100644 index 00000000000..acb1cbb3062 --- /dev/null +++ b/specs/binding-mqtt.spec/src/main/scripts/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/session.will.message.header.exceeds.window/server.rpt @@ -0,0 +1,42 @@ +# +# 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. +# + +accept "zilla://streams/net0" + option zilla:window 8192 + option zilla:transmission "duplex" + option zilla:byteorder "network" + +accepted +connected + +read [0x10 0x2f] # CONNECT + [0x00 0x04] "MQTT" # protocol name + [0x05] # protocol version + [0x06] # flags = will flag, clean start + [0x00 0x0a] # keep alive = 10s + [0x05] # properties + [0x27] 33792 # maximum packet size = 33792 + [0x00 0x03] "one" # client id + [0x00] # will properties + [0x00 0x09] "wills/one" # will topic + [0x00 0x0c] "will payload" # will payload + +write [0x20 0x03] # CONNACK + [0x00] # flags = none + [0x00] # reason code + [0x00] # properties = none + +read aborted diff --git a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java index 383ee934b7a..0e35d62c2f6 100644 --- a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java +++ b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/application/SessionIT.java @@ -119,6 +119,15 @@ public void shouldSendWillMessage10k() throws Exception k3po.finish(); } + @Test + @Specification({ + "${app}/session.will.message.header.exceeds.window/client", + "${app}/session.will.message.header.exceeds.window/server"}) + public void shouldDeferWillMessageWhenHeaderExceedsSessionWindow() throws Exception + { + k3po.finish(); + } + @Test @Specification({ "${app}/session.subscribe/client", diff --git a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java index 00368cb4a7b..10acb33ecb1 100644 --- a/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java +++ b/specs/binding-mqtt.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/streams/network/v5/SessionIT.java @@ -129,6 +129,15 @@ public void shouldConnectWithWillMessageThenNormalDisconnect() throws Exception k3po.finish(); } + @Test + @Specification({ + "${net}/session.will.message.header.exceeds.window/client", + "${net}/session.will.message.header.exceeds.window/server"}) + public void shouldDeferWillMessageWhenHeaderExceedsSessionWindow() throws Exception + { + k3po.finish(); + } + // [MQTT-3.1.2-15] @Test @Specification({ From 627c8ec8ca727b7d036cecc09014a8cc001d46e0 Mon Sep 17 00:00:00 2001 From: Sascha Freiheit Date: Tue, 25 Aug 2026 16:01:38 +0200 Subject: [PATCH 2/2] fix(binding-mqtt-kafka): guard doKafkaData against unattached kafka stream KafkaSessionStream.kafka is only assigned inside a subclass's doKafkaBegin override, so it stays null until that handshake step runs. Every other outbound method on the class (doKafkaEnd, doKafkaAbort, doKafkaReset, doKafkaWindow) already guards on kafka != null, but the three doKafkaData overloads did not, so any caller reaching them first (e.g. sendWillSignal via onMqttAbort) hit an NPE in the shared doData helper's unconditional receiver.accept(...). This is reachable in normal operation: a client that negotiates publishQosMax == 2 with Will + Clean Start, then aborts before doCreateSessionStream has run, leaves the session's kafka field unset when onMqttAbort fires sendWillSignal. Add the same kafka != null guard to all three doKafkaData overloads, at the single shared point that covers sendWillSignal and every other caller. Fixes #2428 Co-Authored-By: Claude Sonnet 5 --- .../stream/MqttKafkaSessionFactory.java | 42 ++++++++------- .../stream/MqttKafkaSessionProxyIT.java | 11 ++++ .../client.rpt | 49 ++++++++++++++++++ .../server.rpt | 51 +++++++++++++++++++ .../client.rpt | 38 ++++++++++++++ .../server.rpt | 35 +++++++++++++ .../binding/mqtt/kafka/streams/KafkaIT.java | 9 ++++ .../binding/mqtt/kafka/streams/MqttIT.java | 11 ++++ 8 files changed, 229 insertions(+), 17 deletions(-) create mode 100644 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 create mode 100644 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 create mode 100644 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 create mode 100644 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 diff --git a/runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java b/runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java index c66902598e3..b4cac2be6e0 100644 --- a/runtime/binding-mqtt-kafka/src/main/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java +++ b/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( diff --git a/runtime/binding-mqtt-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionProxyIT.java b/runtime/binding-mqtt-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionProxyIT.java index ffbe9521ea9..939f667c22b 100644 --- a/runtime/binding-mqtt-kafka/src/test/java/io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionProxyIT.java +++ b/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") diff --git a/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 b/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 new file mode 100644 index 00000000000..2198cb5b8cc --- /dev/null +++ b/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 diff --git a/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 b/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 new file mode 100644 index 00000000000..82182278285 --- /dev/null +++ b/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 diff --git a/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 b/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 new file mode 100644 index 00000000000..f6db14f64c9 --- /dev/null +++ b/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 diff --git a/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 b/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 new file mode 100644 index 00000000000..8c0ca1a99de --- /dev/null +++ b/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 diff --git a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java b/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java index ddacf07e744..f914e7c09c1 100644 --- a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/KafkaIT.java +++ b/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", diff --git a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java b/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java index ae4ee9d6e7f..932cf7fd3ca 100644 --- a/specs/binding-mqtt-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/mqtt/kafka/streams/MqttIT.java +++ b/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",