MQTT CONNECT with a Will message crashes the whole engine (JVM segfault), not just the connection

[sfr-oc]

MQTT CONNECT with a Will message crashes the whole engine (JVM segfault), not just the connection

Zilla version: 2.3.0 (also verified against current develop, i.e. 2.4.0-rc19 — the relevant code paths are byte-identical, so this is not fixed there either)

What happened

Any MQTT client that connects with a Last Will and Testament set brings the whole Zilla process down, not just its own connection. We're running the mqtt → mqtt-kafka proxy chain behind TLS, and this reproduces every single time, with any will topic/payload, as far as we can tell.

Repro is as simple as:

mosquitto_sub -h <host> -p 8883 -v \
  --cafile ca.pem --cert client.pem --key client.key \
  -t "devices/device-123/commands" \
  --will-topic 'devices/device-123/status' \
  --will-payload '{"status":"offline"}'

As soon as the CONNECT (with the will) is decoded, the engine worker throws, self-terminates, and then the whole JVM segfaults and the container restarts — so every other connection on that worker gets dropped too.

Config (anonymized, otherwise matches our setup)

---
name: mqtt-kafka-proxy

vaults:
  client_vault:
    type: filesystem
    options:
      trust:
        store: /path/to/truststore.jks
        type: pkcs12
        password: ${{env.TRUSTSTORE_PASSWORD}}
      keys:
        store: /path/to/keystore.jks
        type: pkcs12
        password: ${{env.KEYSTORE_PASSWORD}}

stores:
  mqtt_store:
    type: memory

bindings:
  north_tcp_server:
    type: tcp
    kind: server
    options:
      host: 0.0.0.0
      port:
        - 8883
    exit: north_tls_server

  north_tls_server:
    type: tls
    kind: server
    vault: client_vault
    options:
      keys:
        - mqtt.example.com
      trust:
        - root-ca
      mutual: required
    exit: north_mqtt_server

  north_mqtt_server:
    type: mqtt
    kind: server
    options:
      store: mqtt_store
    exit: north_mqtt_kafka_mapping

  north_mqtt_kafka_mapping:
    type: mqtt-kafka
    kind: proxy
    options:
      topics:
        sessions: mqtt-sessions
        messages: device-events
        retained: mqtt-retained
      clients:
        - devices/{identity}/#
    routes:
      - when:
          - publish:
              - topic: devices/device-123/status
        with:
          messages: device-events
        exit: north_kafka_cache_client
      - when:
          - publish:
              - topic: devices/device-123/commands
        with:
          messages: device-events
        exit: north_kafka_cache_client
    exit: north_kafka_cache_client

  north_kafka_cache_client:
    type: kafka
    kind: cache_client
    exit: north_kafka_cache_server

  north_kafka_cache_server:
    type: kafka
    kind: cache_server
    options:
      bootstrap:
        - mqtt-retained
        - mqtt-sessions
        - device-events
    exit: north_kafka_client

  north_kafka_client:
    type: kafka
    kind: client
    options:
      servers:
        - broker1:9092
        - broker2:9092
        - broker3:9092
    exit: north_tls_client

  north_tls_client:
    type: tls
    kind: client
    vault: client_vault
    options:
      trust:
        - root-ca
      keys:
        - mqtt.example.com
    exit: north_tcp_client

  north_tcp_client:
    type: tcp
    kind: client

Log / stack trace

engine:events [.../...] ENGINE_STARTED Engine Started.
org.agrona.concurrent.AgentTerminationException: java.lang.IndexOutOfBoundsException: offset=50 is beyond maxLimit=-339
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:961)
        at org.agrona/org.agrona.concurrent.AgentRunner.doWork(AgentRunner.java:304)
        at org.agrona/org.agrona.concurrent.AgentRunner.workLoop(AgentRunner.java:296)
        at org.agrona/org.agrona.concurrent.AgentRunner.run(AgentRunner.java:162)
        at java.base/java.lang.Thread.run(Thread.java:1474)
Caused by: java.lang.IndexOutOfBoundsException: offset=50 is beyond maxLimit=-339
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.types.Flyweight.wrap(Flyweight.java:59)
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.types.OctetsFW.wrap(OctetsFW.java:39)
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer.onDecodeConnectWillPayload(MqttServerFactory.java:3593)
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory.decodeConnectWillMessagePayload(MqttServerFactory.java:1251)
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer.decodeNetwork(MqttServerFactory.java:5359)
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer.decodeNetwork(MqttServerFactory.java:5341)
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer$MqttSessionStream.onSessionWindow(MqttServerFactory.java:5688)
        at io.aklivity.zilla.runtime.binding.mqtt@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.internal.stream.MqttServerFactory$MqttServer$MqttSessionStream.onSession(MqttServerFactory.java:5648)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleReadInitial(EngineWorker.java:1544)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.handleRead(EngineWorker.java:1477)
        at io.aklivity.zilla.runtime.common.agrona@2.3.0/io.aklivity.zilla.runtime.common.agrona.concurrent.ManyToOneRingBuffer.readEx(ManyToOneRingBuffer.java:316)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doWork(EngineWorker.java:947)
        ... 4 more
java.lang.NullPointerException: Cannot invoke "io.aklivity.zilla.runtime.engine.binding.function.MessageConsumer.accept(int, io.aklivity.zilla.runtime.common.agrona.buffer.DirectBufferEx, int, int)" because "receiver" is null
        at io.aklivity.zilla.runtime.binding.mqtt.kafka@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSessionFactory.doData(MqttKafkaSessionFactory.java:4636)
        at io.aklivity.zilla.runtime.binding.mqtt.kafka@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSessionFactory$KafkaSessionStream.doKafkaData(MqttKafkaSessionFactory.java:3063)
        at io.aklivity.zilla.runtime.binding.mqtt.kafka@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSessionFactory$KafkaSessionStream.sendWillSignal(MqttKafkaSessionFactory.java:3235)
        at io.aklivity.zilla.runtime.binding.mqtt.kafka@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSessionFactory$MqttSessionProxy.onMqttAbort(MqttKafkaSessionFactory.java:937)
        at io.aklivity.zilla.runtime.binding.mqtt.kafka@2.3.0/io.aklivity.zilla.runtime.binding.mqtt.kafka.internal.stream.MqttKafkaSessionFactory$MqttSessionProxy.onMqttMessage(MqttKafkaSessionFactory.java:538)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.doSyntheticAbort(EngineWorker.java:1993)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.lambda$detachStreams$0(EngineWorker.java:630)
        at java.base/java.lang.Iterable.forEach(Iterable.java:75)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.detachStreams(EngineWorker.java:623)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.NamespaceRegistry.detachBinding(NamespaceRegistry.java:288)
        at java.base/java.lang.Iterable.forEach(Iterable.java:75)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.NamespaceRegistry.detach(NamespaceRegistry.java:150)
        at org.agrona/org.agrona.collections.Int2ObjectHashMap$ValueCollection.forEach(Int2ObjectHashMap.java:1328)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineRegistry.detachAll(EngineRegistry.java:197)
        at io.aklivity.zilla.runtime.engine@2.3.0/io.aklivity.zilla.runtime.engine.internal.registry.EngineWorker.onClose(EngineWorker.java:1078)
        at org.agrona/org.agrona.concurrent.AgentRunner.run(AgentRunner.java:166)
        at java.base/java.lang.Thread.run(Thread.java:1474)
#
# A fatal error has been detected by the Java Runtime Environment:
#
#  SIGSEGV (0xb) at pc=0x00007f6c20c4d02d, pid=1, tid=2
# Problematic frame:
# J 4894 c2 jdk.internal.misc.Unsafe.getLongUnaligned(Ljava/lang/Object;JZ)J java.base@25.0.3
engine:events [.../...] ENGINE_STOPPED Engine Stopped.

(JRE: Temurin 25.0.3+9, mixed mode, g1 gc, linux-amd64)

What I think is going on (happy to be corrected)

I went digging in MqttServerFactory and think I found it. In onDecodeConnectWillPayload (around line 3591 on develop):

int payloadSize = Math.min(limit - offset, session.initialBudget() - headerSize);
...
final OctetsFW payload = payloadRO.wrap(buffer, offset, offset + payloadSize);

session here is the internal stream to the mqtt-kafka proxy side. session.initialBudget() is initialMax - (initialSeq-initialAck) - initialPad, i.e. whatever window the mqtt-kafka side has granted back via WINDOW. The session stream is opened (doEstablishSession) in the very same decode pass that then goes on to decode the will payload, so there's no way a WINDOW frame for it could have arrived yet — initialMax is still 0 at this point. So session.initialBudget() ends up <= 0, and since headerSize (topic, delay, qos, flags, properties, etc.) is always > 0, session.initialBudget() - headerSize is basically guaranteed to be negative. That makes payloadSize negative, which makes offset + payloadSize end up less than offset, and OctetsFW.wrap() throws because it requires limit >= offset — which matches the numbers in the trace (offset=50, maxLimit=-339).

That's the initial crash. It kills the engine worker thread. What makes it fatal for the whole process rather than just that worker is a second, separate bug: while the worker tears itself down, it synthetically aborts the still-open MQTT session, which calls sendWillSignal() → doKafkaData() → doData() in MqttKafkaSessionFactory, and that does receiver.accept(...) without checking whether receiver is actually set — in our case it's null because the Kafka-side session stream had never been fully established. That second, uncaught NPE seems to be what tips this over into the JVM segfault instead of a clean worker restart.

I noticed #1809 addressed a very similar-sounding class of problem (uncaught binding exception → engine worker self-terminates → JVM crash via drain() touching unmapped memory) and that fix is already in 2.3.0 and current develop. This looks like a related but distinct gap that #1809 doesn't cover, since the NPE here happens earlier, inside onClose()'s own synthetic-abort loop, before any buffers get released.

Expected behavior

A client connecting with a will message shouldn't be able to take down the whole engine.

Reproducibility

100% for us — every CONNECT with a will message on this proxy chain triggers it.

We're happy to open a PR for this (we've already dug into both spots above), just wanted to get it tracked first.

[sfr-oc]

We've built our fix into a new Docker container and rolled it out. For us, the fix works flawlessly.

[jfallows]

@sfr-oc thank you for filing this issue and implementing the fix.

Please open a PR so we can review and merge into a patch release for you.

[sfr-oc]

it is open:
#2429