explicitly close and recreat dead kafka consumers

Author: maldwg

config.yaml

@@ -278,26 +278,30 @@ pipeline:
     plugins: []
   monitoring:
     kafka_consumer:
-      # Monitoring produces several records per pipeline message. Consume them
-      # in larger blocks so schema decoding, ClickHouse writes, and offset
-      # commits are amortized across the batch.
-      batch_size: 5000
+      # Keep the uncommitted replay window bounded. ClickHouse already flushes
+      # each table every 50 rows, so a much larger Kafka batch does not create
+      # larger ClickHouse inserts; it only delays the Kafka offset commit.
+      batch_size: 500
       # Match the ClickHouse partial-batch window. A shorter timeout forces
       # sparse tables into many small synchronous inserts.
       timeout_ms: 2000
     clickhouse_connector:
       batch_size: 50  # do not set higher
       batch_timeout: 2.0
+      # Keep a sink outage well below Kafka's max-poll interval. The monitoring
+      # agent leaves its consumer group before longer reconnect attempts.
+      connect_timeout_seconds: 10
+      operation_timeout_seconds: 30
 
 
   zeek:
     sensors:
       zeek:
-        static_analysis: true
+        static_analysis: false
         protocols:
           - dns
         interfaces:
-          - eth1
+          - wlp3s0
 
 environment:
   kafka_brokers:
@@ -317,13 +321,21 @@ environment:
     # Allow long-running detector batches without Kafka forcing a group rebalance.
     # Default librdkafka value is 300000 ms (5 minutes)
     max_poll_interval_ms: 300000
+    # A commit belongs to one consumer generation. Give transient coordinator
+    # errors a short recovery window, then resubscribe instead of retrying a
+    # stale generation forever.
+    commit_retry_timeout_seconds: 30
   kafka_producer:
     # Compress JSON pipeline and monitoring batches before replication.
     compression_type: zstd
+    # Bound native transactional calls so a broker/DNS outage returns control
+    # to coordinated producer/consumer recovery.
+    transaction_api_timeout_seconds: 30
+    message_timeout_ms: 60000
   kafka_transaction_batch:
     # Global fallback. Exact topic settings override stage settings, which
     # override these defaults. Environment variables remain final overrides.
-    size: 250
+    size: 50
     timeout_ms: 50
     stages:
       # CPU-heavy inference must return to Kafka much more frequently.

docs/configuration.rst

@@ -602,6 +602,18 @@ To entirely skip the anomaly detection phase, you can set ``inspector_module_nam
    * - clickhouse_connector.batch_timeout
      - ``2.0``
      - Maximum time in seconds before a partial monitoring batch is written.
+   * - clickhouse_connector.connect_timeout_seconds
+     - ``10``
+     - Timeout for one ClickHouse connection attempt. Reconnection continues after the monitoring consumer has left its Kafka group.
+   * - clickhouse_connector.operation_timeout_seconds
+     - ``30``
+     - Timeout for one ClickHouse HTTP request. A failed insert is replayed from uncommitted Kafka offsets after dependency recovery.
+   * - kafka_consumer.batch_size
+     - ``500``
+     - Maximum monitoring Kafka records fetched in one explicit sink batch.
+   * - kafka_consumer.timeout_ms
+     - ``2000``
+     - Maximum time spent collecting a partial monitoring Kafka batch.
 
 ``pipeline.zeek``
 ^^^^^^^^^^^^^^^^^
@@ -649,18 +661,27 @@ The following parameters control the infrastructure of the software.
    * - kafka_consumer.max_poll_interval_ms
      - ``1800000`` fallback; ``300000`` in the supplied ``config.yaml``
      - Maximum time in milliseconds between Kafka consumer polls before Kafka removes the consumer from its group. Increase this for long-running detector batches and keep batch processing comfortably below the interval. See :doc:`kafka_recovery` for automatic membership recovery, rebalance handling, and delivery guarantees.
+   * - kafka_consumer.commit_retry_timeout_seconds
+     - ``30``
+     - Maximum time to retry transient offset-commit failures with one consumer generation. Membership errors trigger immediate resubscription instead. ``KAFKA_CONSUMER_COMMIT_RETRY_TIMEOUT_SECONDS`` is the final environment override.
    * - kafka_transaction_batch.size / timeout_ms
-     - ``250`` / ``50`` in the supplied ``config.yaml``
+     - ``50`` / ``50`` in the supplied ``config.yaml``
      - Global Kafka consume/transaction batch defaults. ``size`` counts Kafka records and ``timeout_ms`` controls how long a consumer waits while collecting a batch.
    * - kafka_transaction_batch.stages
      - Detector ``size: 10``; alerter ``size: 25``
      - Per-consuming-stage overrides. Short names such as ``detector`` and fully-qualified names such as ``data_analysis.detector`` are supported.
    * - kafka_transaction_batch.topics
-     - Domainator input topic ``size: 5``
+     - None by default
      - Per-topic overrides keyed by the complete consumed topic name. Exact-topic settings override stage settings, which override the global defaults. See :doc:`kafka_recovery` for examples and environment-variable precedence.
    * - kafka_producer.compression_type
      - ``zstd``
      - Compression codec used by HAMSTRING's Python Kafka producers. Zeek is an external producer and must be configured separately if its image supports compression settings.
+   * - kafka_producer.transaction_api_timeout_seconds
+     - ``30``
+     - Timeout for one transactional producer initialization, commit, or abort call. A timeout causes coordinated producer and consumer recovery. ``KAFKA_TRANSACTION_API_TIMEOUT_SECONDS`` is the final environment override.
+   * - kafka_producer.message_timeout_ms
+     - ``60000``
+     - Maximum delivery time for a transactional Kafka output record. ``KAFKA_PRODUCER_MESSAGE_TIMEOUT_MS`` is the final environment override.
    * - kafka_topics.replication_factor
      - ``3``
      - Replication factor used when creating new Kafka topics. At runtime this is capped to the number of configured Kafka brokers.

docs/kafka_recovery.rst

@@ -41,6 +41,32 @@ stages.
 HAMSTRING also handles librdkafka's local ``_MAX_POLL_EXCEEDED`` consumer error
 through the same consumer-reset path.
 
+Broker and Docker-network outages
+---------------------------------
+
+Kafka transport failures such as ``_TIMED_OUT``, ``_MSG_TIMED_OUT``,
+``_TRANSPORT``, and ``_RESOLVE`` use coordinated recovery. Native transaction
+initialization, commit, and abort calls have an explicit API timeout. When a
+transactional stage loses Kafka connectivity it:
+
+#. aborts the transaction within the configured API timeout;
+#. closes the consumer, thereby leaving its current group generation;
+#. recreates and initializes the transactional producer;
+#. subscribes a fresh consumer after Kafka is reachable;
+#. reconsumes the uncommitted input batch.
+
+The service may remain in a visible reconnect loop while Kafka or Docker DNS
+is unavailable, but it no longer holds stale consumer membership while doing
+so. ``_PURGE_QUEUE`` delivery callbacks during the abort describe records
+discarded from the old local producer. They do not mean that the corresponding
+input offsets were committed.
+
+``environment.kafka_producer.transaction_api_timeout_seconds`` bounds each
+native transactional API call. ``message_timeout_ms`` bounds how long produced
+records may remain undelivered. The environment variables
+``KAFKA_TRANSACTION_API_TIMEOUT_SECONDS`` and
+``KAFKA_PRODUCER_MESSAGE_TIMEOUT_MS`` are final deployment overrides.
+
 Rebalance tracking
 ------------------
 
@@ -94,7 +120,7 @@ the consuming pipeline stage, or by the complete consumed topic name:
 
    environment:
      kafka_transaction_batch:
-       size: 250
+       size: 50
        timeout_ms: 50
        stages:
          detector:
@@ -120,9 +146,9 @@ corresponding short name. Exact topic keys must contain the full topic name
 consumed by the worker. If one consumer subscribes to multiple topics, it uses
 the smallest effective size and timeout across those topics.
 
-The supplied configuration sets the detector stage to 10 records, the alerter
-stage to 25 records, and the initial Domainator topic to 5 records. Other stages
-inherit the global value. A caller that explicitly supplies ``max_messages`` or
+The supplied configuration sets the global value to 50 records, the detector
+stage to 10 records, and the alerter stage to 25 records. Other stages inherit
+the global value. A caller that explicitly supplies ``max_messages`` or
 ``timeout_ms`` to ``consume_batch`` still overrides the resolved defaults; the
 collector uses this for its application-level batching configuration.
 
@@ -133,6 +159,38 @@ Operationally, alert on:
 * transaction aborts while consumer lag increases;
 * no successful output transaction while input offsets continue to arrive.
 
+Monitoring sink recovery
+------------------------
+
+The monitoring agent is not part of a Kafka transaction: it inserts a consumed
+batch into ClickHouse and then commits its Kafka offsets. Its recovery guarantee
+is therefore **at least once**, not exactly once.
+
+The supplied configuration consumes at most 500 monitoring records per Kafka
+batch while ClickHouse continues to flush each table every 50 rows. Increasing
+the Kafka value does not increase the ClickHouse insert size; it only delays the
+offset commit and enlarges the batch that must be replayed after a failure.
+
+ClickHouse connection and request calls are bounded by
+``pipeline.monitoring.clickhouse_connector.connect_timeout_seconds`` and
+``operation_timeout_seconds``. If an insert fails, the agent closes its Kafka
+consumer before waiting for ClickHouse to return. It discards the local batch,
+reconnects ClickHouse, subscribes a fresh Kafka consumer, and replays records
+from committed offsets.
+
+Normal offset commits retry transient coordinator errors only for
+``environment.kafka_consumer.commit_retry_timeout_seconds``. Membership errors
+such as ``UNKNOWN_MEMBER_ID`` skip that retry window and immediately cause a
+fresh subscription. Retrying a commit with the same expired generation can
+never succeed.
+
+An outage can occur after ClickHouse accepted an insert but before Kafka
+accepted its offset commit. Replaying the Kafka record can consequently insert
+a duplicate monitoring row. This is preferable to committing an offset for a
+row that may not have reached ClickHouse. Strict duplicate prevention requires
+an idempotent ClickHouse schema keyed by the source Kafka topic, partition, and
+offset; the current raw monitoring tables do not provide that guarantee.
+
 Delivery guarantees outside Kafka
 ---------------------------------
 

src/base/kafka/__init__.py

@@ -8,6 +8,7 @@
 )
 from src.base.kafka.errors import (
     KafkaConsumerMembershipLost,
+    KafkaInfrastructureUnavailable,
     KafkaMessageFetchException,
     TooManyFailedAttemptsError,
 )
@@ -31,6 +32,7 @@
     "ExactlyOnceKafkaProduceHandler",
     "KafkaConsumeHandler",
     "KafkaConsumerMembershipLost",
+    "KafkaInfrastructureUnavailable",
     "KafkaHandler",
     "KafkaMessageFetchException",
     "KafkaProduceHandler",

src/base/kafka/config.py

@@ -18,11 +18,29 @@
 KAFKA_CONSUMER_MAX_POLL_INTERVAL_MS = int(
     KAFKA_CONSUMER_CONFIG.get("max_poll_interval_ms", 1_800_000)
 )
+KAFKA_CONSUMER_COMMIT_RETRY_TIMEOUT_SECONDS = float(
+    os.getenv(
+        "KAFKA_CONSUMER_COMMIT_RETRY_TIMEOUT_SECONDS",
+        KAFKA_CONSUMER_CONFIG.get("commit_retry_timeout_seconds", 30),
+    )
+)
 KAFKA_PRODUCER_CONFIG = CONFIG["environment"].get("kafka_producer", {})
 KAFKA_PRODUCER_COMPRESSION_TYPE = os.getenv(
     "KAFKA_PRODUCER_COMPRESSION_TYPE",
     KAFKA_PRODUCER_CONFIG.get("compression_type", "zstd"),
 )
+KAFKA_TRANSACTION_API_TIMEOUT_SECONDS = float(
+    os.getenv(
+        "KAFKA_TRANSACTION_API_TIMEOUT_SECONDS",
+        KAFKA_PRODUCER_CONFIG.get("transaction_api_timeout_seconds", 30),
+    )
+)
+KAFKA_PRODUCER_MESSAGE_TIMEOUT_MS = int(
+    os.getenv(
+        "KAFKA_PRODUCER_MESSAGE_TIMEOUT_MS",
+        KAFKA_PRODUCER_CONFIG.get("message_timeout_ms", 60_000),
+    )
+)
 
 KAFKA_TRANSACTION_BATCH_CONFIG = CONFIG["environment"].get(
     "kafka_transaction_batch", {}

src/base/kafka/consumer.py

@@ -12,6 +12,7 @@
 from src.base.kafka.client import KafkaHandler
 from src.base.kafka.errors import (
     KafkaConsumerMembershipLost,
+    KafkaInfrastructureUnavailable,
     KafkaMessageFetchException,
     TooManyFailedAttemptsError,
 )
@@ -28,7 +29,7 @@
     topic_partition_count,
 )
 from src.base.log_config import get_logger
-from src.base.retry import retry_forever
+from src.base.retry import retry_forever, retry_with_timeout
 
 logger = get_logger()
 
@@ -51,6 +52,16 @@ def __init__(self, topics: str | list[str], stage: str | None = None) -> None:
         self.brokers = kafka_config.bootstrap_servers()
         self.conf = self._build_consumer_conf()
         self._connect_consumer()
+        logger.info(
+            "Kafka consumer configured for stage %s with batch size %d, "
+            "batch timeout %dms, max poll interval %dms, and commit retry "
+            "timeout %.1fs.",
+            self.stage or "unspecified",
+            self.transaction_batch_size,
+            self.transaction_batch_timeout_ms,
+            kafka_config.KAFKA_CONSUMER_MAX_POLL_INTERVAL_MS,
+            kafka_config.KAFKA_CONSUMER_COMMIT_RETRY_TIMEOUT_SECONDS,
+        )
 
     def _build_consumer_conf(self) -> dict:
         return {
@@ -95,6 +106,15 @@ def connect():
         )
 
     def _reset_consumer(self) -> None:
+        self.disconnect_for_recovery()
+        self.reconnect_after_recovery()
+
+    def disconnect_for_recovery(self, reason: Exception | None = None) -> None:
+        """Leave the current group before waiting for a failed dependency."""
+        if reason:
+            logger.warning(
+                "Closing Kafka consumer before dependency recovery: %s", reason
+            )
         try:
             if self.consumer:
                 self.consumer.close()
@@ -103,7 +123,14 @@ def _reset_consumer(self) -> None:
                 "Ignoring Kafka consumer close failure during reconnect: %s",
                 exception,
             )
+        self.consumer = None
         self._last_consumed_message = None
+        self._assignment_epoch += 1
+        self._assignment_valid = False
+        self._assigned_partitions.clear()
+
+    def reconnect_after_recovery(self) -> None:
+        """Create and subscribe a new consumer after dependencies are reachable."""
         self._connect_consumer()
 
     def recover_group_membership(self, reason: Exception | None = None) -> None:
@@ -181,29 +208,55 @@ def commit(
     ) -> None:
         """Commit an explicit batch or the latest record from ``consume``."""
         if not self.consumer:
-            return
+            raise KafkaInfrastructureUnavailable(
+                "Kafka offset commit requested without a connected consumer."
+            )
+
+        def commit_with_current_membership(operation) -> None:
+            def commit_once():
+                try:
+                    return operation()
+                except KafkaException as exception:
+                    if is_consumer_membership_error(
+                        exception.args[0] if exception.args else None
+                    ):
+                        raise KafkaConsumerMembershipLost(
+                            f"Kafka offset commit used stale membership: {exception}"
+                        ) from exception
+                    raise
+
+            try:
+                retry_with_timeout(
+                    commit_once,
+                    "Kafka consumer batch offset commit",
+                    kafka_config.RETRY_SETTINGS,
+                    kafka_config.KAFKA_CONSUMER_COMMIT_RETRY_TIMEOUT_SECONDS,
+                    retryable=(KafkaException, RuntimeError, OSError),
+                )
+            except KafkaConsumerMembershipLost:
+                raise
+            except (KafkaException, RuntimeError, OSError) as exception:
+                raise KafkaInfrastructureUnavailable(
+                    "Kafka offset commit did not recover before its retry "
+                    f"deadline: {exception}"
+                ) from exception
 
         if consumed_messages is not None:
             if not consumed_messages:
                 return
-            retry_forever(
+            self.ensure_batch_assignment_current(consumed_messages)
+            commit_with_current_membership(
                 lambda: self.consumer.commit(
                     offsets=self.offsets_for(consumed_messages),
                     asynchronous=False,
-                ),
-                "Kafka consumer batch offset commit",
-                kafka_config.RETRY_SETTINGS,
-                retryable=(KafkaException, RuntimeError, OSError),
+                )
             )
             self._last_consumed_message = None
             return
 
         if self._last_consumed_message is not None:
-            retry_forever(
-                lambda: self.consumer.commit(self._last_consumed_message),
-                "Kafka consumer offset commit",
-                kafka_config.RETRY_SETTINGS,
-                retryable=(KafkaException, RuntimeError, OSError),
+            commit_with_current_membership(
+                lambda: self.consumer.commit(self._last_consumed_message)
             )
             self._last_consumed_message = None
 

src/base/kafka/errors.py

@@ -11,3 +11,7 @@ class KafkaMessageFetchException(Exception):
 
 class KafkaConsumerMembershipLost(Exception):
     """Raised when a transaction uses stale consumer-group membership."""
+
+
+class KafkaInfrastructureUnavailable(Exception):
+    """Raised when Kafka transport failure requires coordinated client recovery."""