Troubleshooting AIS Kafka & MQTT Connector Failures

Problem Framing #

An AIS ingestion connector rarely fails with a stack trace. It fails silently: the Real-Time AIS Stream Ingestion Pipelines dashboard shows a flat message rate, consumer lag climbs, and downstream Parquet partitions stop appearing — yet the process is still running and its logs look calm. Whether the feed arrives over an MQTT broker fed by a shore-based VHF receiver or a Kafka topic hydrated from a satellite AIS provider, the symptom is the same: bytes stop flowing while the connector reports “connected”. This page is the diagnostic playbook for that failure class. It classifies the root causes, reproduces each from real broker log lines, and rebuilds the subscriber so it recovers on its own instead of hanging until an operator restarts it at 03:00.

The failure taxonomy below groups the causes by the layer they live in — transport, authentication, session, and delivery-semantics — because the fix for each layer is different and applying the wrong one (for example, bumping session.timeout.ms when the real problem is a shared MQTT client-id) wastes an on-call rotation.

Root Cause: Reading the Stall From the Logs #

The single most useful skill is mapping a log line to a layer. Three signatures cover the majority of AIS connector stalls.

MQTT rc=5 (not authorized). paho-mqtt reports the CONNACK return code in on_connect. A code of 5 means the broker rejected the credentials or the ACL forbids the requested topic filter. The subscriber’s loop_forever() keeps spinning, so nothing crashes — it simply never receives a message:

ais-mqtt  INFO  connected with result code 5
ais-mqtt  INFO  subscribed to ais/nmea/#
# ...then silence. rc=5 means the SUBSCRIBE was never honoured.

Kafka _TRANSPORT / Disconnected. librdkafka surfaces transport faults through the error callback, not as an exception. A TLS or DNS problem shows up as a broker marked DOWN and an endless reconnect cycle:

%3|rdkafka#consumer-1| FAIL [thrd:sasl_ssl://broker:9093]: broker:9093: Disconnected (after 12ms in state UP)
%3|rdkafka#consumer-1| ERROR: 1/1 brokers are down

Subscribe succeeds but no messages. The hardest case: rc=0, subscription acknowledged, zero payloads. This is almost always a shared client-id — two subscribers with the same client_id repeatedly evict each other, or an upstream QoS0 publisher dropping messages during a receiver reconnect. Because MQTT permits exactly one session per client-id, the second connection forcibly disconnects the first, and the two processes thrash forever.

Reproduce the shared-client-id thrash locally:

# Two subscribers, SAME client-id — they will kick each other in a loop.
import paho.mqtt.client as mqtt  # paho-mqtt==2.1.0

def make(cid: str) -> mqtt.Client:
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=cid)
    c.connect("test.mosquitto.org", 1883, keepalive=30)
    return c

a = make("ais-shore-01")
b = make("ais-shore-01")   # BUG: duplicate id — broker disconnects `a`

Step-by-Step Fix #

Step 1 — Probe the Kafka broker before touching the consumer #

Never debug a stalled Kafka consumer by reading its business logic first. Confirm the broker is reachable and the group is live. This probe raises on failure instead of logging and continuing, so it is safe to run as a pre-flight gate in the same container:

import logging
from confluent_kafka.admin import AdminClient
from confluent_kafka import KafkaException

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s — %(message)s")
logger = logging.getLogger("ais.probe")


def assert_broker_reachable(bootstrap_servers: str, timeout_s: float = 10.0) -> None:
    """Raise KafkaException if no broker in the cluster answers a metadata request."""
    admin = AdminClient({"bootstrap.servers": bootstrap_servers})
    md = admin.list_topics(timeout=timeout_s)
    if not md.brokers:
        raise KafkaException(f"No brokers reachable at {bootstrap_servers}")
    for bid, broker in md.brokers.items():
        logger.info("broker %s up at %s:%d", bid, broker.host, broker.port)

A KafkaException: ... Failed to acquire metadata here isolates the fault to transport or auth immediately — it is not a consumer-logic bug, and the AIS Kafka consumer code is exonerated before you open it.

Step 2 — Rebuild the MQTT subscriber to survive drops #

The most common shore-feed problem is a subscriber that connects once and never recovers from a network blip. The fix is threefold: a long-enough keepalive, clean_session=False so the broker queues messages during a brief disconnect, and explicit reconnect backoff in on_disconnect. paho-mqtt 2.x already retries via loop_forever(), but the defaults are too aggressive for a satellite backhaul with 800 ms RTT.

from __future__ import annotations

import logging
import time
from dataclasses import dataclass

import paho.mqtt.client as mqtt  # paho-mqtt==2.1.0

logger = logging.getLogger("ais.mqtt")


@dataclass(frozen=True)
class MqttConfig:
    host: str
    port: int = 8883          # TLS
    keepalive: int = 60       # seconds; must exceed the feed's quiet gaps
    topic: str = "ais/nmea/#"
    qos: int = 1              # at-least-once; QoS0 silently drops on reconnect
    client_id: str = "ais-shore-ingest-01"


class ResilientAisSubscriber:
    """A paho MQTT subscriber that reconnects with capped exponential backoff
    and refuses to run with a QoS0 + clean_session combination that loses data."""

    def __init__(self, cfg: MqttConfig) -> None:
        if cfg.qos == 0 and cfg.keepalive > 0:
            logger.warning("QoS0 selected — messages during any reconnect are lost")
        self._cfg = cfg
        self._backoff = 1.0
        self._backoff_max = 60.0
        # persistent session (clean_session=False) so the broker buffers
        # QoS1 messages while we are briefly offline
        self._client = mqtt.Client(
            mqtt.CallbackAPIVersion.VERSION2,
            client_id=cfg.client_id,
            clean_session=False,
        )
        self._client.on_connect = self._on_connect
        self._client.on_disconnect = self._on_disconnect
        self._client.on_message = self._on_message
        self._client.tls_set()  # system CA bundle; raises on missing certs

    def _on_connect(self, client: mqtt.Client, _u, _f, rc, _p=None) -> None:
        code = int(rc.value) if hasattr(rc, "value") else int(rc)
        if code != 0:
            # rc=5 -> not authorized; do NOT keep silently retrying forever
            raise ConnectionRefusedError(f"MQTT CONNACK rc={code} for {self._cfg.host}")
        self._backoff = 1.0
        client.subscribe(self._cfg.topic, qos=self._cfg.qos)
        logger.info("connected; subscribed to %s at qos%d", self._cfg.topic, self._cfg.qos)

    def _on_disconnect(self, client: mqtt.Client, _u, _f, rc, _p=None) -> None:
        code = int(rc.value) if hasattr(rc, "value") else int(rc)
        logger.warning("disconnected rc=%d — backing off %.0fs", code, self._backoff)
        time.sleep(self._backoff)
        self._backoff = min(self._backoff * 2.0, self._backoff_max)

    def _on_message(self, _c, _u, msg: mqtt.MQTTMessage) -> None:
        payload = msg.payload.decode("ascii", errors="replace")
        # hand off to the NMEA reassembly / decode stage here
        logger.debug("ais %s: %s", msg.topic, payload)

    def run(self) -> None:
        self._client.connect(self._cfg.host, self._cfg.port, self._cfg.keepalive)
        self._client.loop_forever(retry_first_connection=True)

Two decisions matter most. clean_session=False turns the subscription into a persistent session: while the process is briefly offline, the broker holds QoS1 messages and replays them on reconnect, so a 5-second network blip no longer drops a minute of position reports. And raising ConnectionRefusedError on rc=5 converts a silent stall into a loud, page-able failure instead of an infinite polite retry.

Step 3 — Give every subscriber a unique client-id #

Shared client-ids are the leading cause of “subscribe succeeds, no messages”. Derive the id from the pod/host so two replicas cannot collide:

import os
import socket


def unique_client_id(prefix: str = "ais-ingest") -> str:
    """Stable-per-host, unique-per-process MQTT client id."""
    host = os.environ.get("HOSTNAME") or socket.gethostname()
    return f"{prefix}-{host}-{os.getpid()}"

Step 4 — A liveness gate that raises on zero throughput #

A connector that is “up” but delivering nothing must fail its healthcheck. Track message arrivals in a rolling window and raise when the rate collapses; wire this to a Kubernetes liveness probe so the orchestrator restarts a wedged pod:

import time
import threading
from collections import deque


class ThroughputMonitor:
    """Thread-safe rolling-window message-rate monitor for liveness checks."""

    def __init__(self, window_s: float = 120.0, min_rate_hz: float = 0.5) -> None:
        self._window_s = window_s
        self._min_rate = min_rate_hz
        self._events: deque[float] = deque()
        self._lock = threading.Lock()

    def record(self) -> None:
        now = time.monotonic()
        with self._lock:
            self._events.append(now)
            cutoff = now - self._window_s
            while self._events and self._events[0] < cutoff:
                self._events.popleft()

    def assert_healthy(self) -> None:
        with self._lock:
            rate = len(self._events) / self._window_s
        if rate < self._min_rate:
            raise RuntimeError(
                f"AIS throughput {rate:.3f} Hz below floor {self._min_rate} Hz "
                f"over {self._window_s:.0f}s window — connector stalled"
            )

Verification #

Drive the monitor with a synthetic burst and confirm it flips from healthy to stalled, so you trust the gate before it guards production:

def test_liveness_gate() -> None:
    mon = ThroughputMonitor(window_s=10.0, min_rate_hz=1.0)
    for _ in range(50):        # 50 messages -> 5 Hz, healthy
        mon.record()
    mon.assert_healthy()       # passes

    time.sleep(11)             # let the window drain
    try:
        mon.assert_healthy()
        raise AssertionError("gate failed to detect the stall")
    except RuntimeError:
        pass                   # expected: rate collapsed to 0 Hz
    print("PASS — liveness gate detects throughput collapse")

On the broker side, confirm Kafka lag is bounded and falling after the fix:

kafka-consumer-groups.sh --bootstrap-server broker:9093 \
  --describe --group ais-ingestion-prod
# LAG column must trend toward zero once the consumer is healthy.

Edge Cases and Gotchas #

• QoS vs persistent-session tradeoff. QoS1 with clean_session=False guarantees at-least-once delivery but shifts the deduplication burden downstream — the broker will replay buffered messages after a reconnect, so pair this connector with deduplicating AIS messages in real-time streams or you will double-count positions. QoS0 avoids replay but drops silently on any blip; never use it for a primary feed.
• Firewalled satellite backhaul. S-AIS providers often deliver over a NAT gateway that closes idle TCP flows after ~90 s. If keepalive exceeds that idle timeout, the broker still thinks you are connected while the middlebox has already dropped the flow — a classic half-open stall. Set keepalive below the shortest middlebox idle timeout (30–45 s is safe).
• Broker retention masquerading as data loss. If a Kafka consumer is offline longer than the topic’s retention.ms, the oldest offsets are deleted and auto.offset.reset decides whether you skip ahead (latest) or re-read (earliest). A gap that looks like a connector bug can actually be retention eviction; check retention.ms before blaming the client. Persistent gaps then flow into AIS data quality & gap filling.
• Clock skew makes replayed dupes look distinct. After a reconnect replay, receiver-added arrival timestamps may differ even though the AIS payload is identical — see handling AIS timestamp drift & clock skew before you key any dedup logic on timestamps.

---

Related #

• Deduplicating AIS Messages in Real-Time Streams — required when QoS1 or at-least-once replay reintroduces duplicates after a reconnect
• Handling AIS Timestamp Drift & Clock Skew — why replayed messages can look distinct even when they are duplicates
• Building an AIS Kafka Consumer in Python — the consumer these probes and gates protect

Up: Real-Time AIS Stream Ingestion Pipelines