A broker outage exhausts every pending row's attempt budget; when Kafka returns the relay publishes nothing

[AlexeyShalaev]

Measured on 0.2.0 with aiokafka 0.14.0 and a Kafka container paused mid-run, so the broker neither answers nor refuses. Twenty pending outbox rows, OmniBoxDomainService(max_attempts=6) (the default), KafkaEventPublisher(max_infra_retries=1), a relay loop calling publish_batch once per cycle:

--- Kafka paused; the relay keeps ticking every cycle ---
  cycle 1 at  40.1 s: {'pending/attempts=1': 20}
  cycle 2 at  80.2 s: {'pending/attempts=2': 20}
  cycle 3 at 120.3 s: {'pending/attempts=3': 20}
  cycle 4 at 160.5 s: {'pending/attempts=4': 20}
  cycle 5 at 200.6 s: {'pending/attempts=5': 20}
  cycle 6 at 240.7 s: {'failed/attempts=6': 20}
  cycle 7 at 240.7 s: {'failed/attempts=6': 20}

--- Kafka is back ---
  next relay cycle: {'failed/attempts=6': 20}; messages on the topic: 0
  after requeue_failed on every failed row and one more cycle: {'completed/attempts=0': 20}; messages on the topic: 20

Every cycle the broker is unreachable costs every pending row one attempt, and after max_attempts cycles the whole backlog is failed, which is terminal. When Kafka returns the relay publishes nothing, because fetch_and_lock_pending filters on attempts_made < max_attempts. The events are recovered only by someone running requeue_failed on each row. A relay ticking every second turns a four-minute broker outage into a backlog that never drains on its own, and the outbox's one promise, that nothing is lost, is kept only in the sense that the rows are still in the table.

The attempt budget is the right tool for a row that cannot be published: a payload the broker rejects, a topic that does not exist, a serialization error. It is the wrong tool for a broker that is not there, which is a property of the cycle, not of any row. Rule 13 says nothing classifies errors for the pipeline and that ErrorClassifier is used only inside the Kafka publisher's own infrastructure retry; once max_infra_retries is spent the exception reaches the pipeline as a counted failure like any other.

What I think it needs: infrastructure failures from the broker (connection, metadata, request timeout, the classes ErrorClassifier already knows) should reach the outbox pipeline as a non-counted failure with a next_retry_at backoff, the way handler_retry(count_as_attempt=False, next_retry_at=…) already lets a handler say "not my fault, try later". The publisher step is the natural place: it has the exception, it has the classifier, and the pipeline already supports the outcome. A counted failure stays reserved for errors that are about the event. With that, the table above ends with twenty pending rows that publish on the first cycle after Kafka returns, and requeue_failed goes back to being an operator's tool for genuinely poisoned rows. The docs (rule 13, the outbox guide, the runbook if there is one) follow.

Script: kafka_down_lab.py in https://github.com/bedrock-python/bedrock-python.github.io/tree/docs/production-python-series/docs/blog/lab/2026-09-07-when-kafka-is-down. It takes about five minutes because the paused cycles wait on aiokafka's request timeout per event.

[AlexeyShalaev]

I ran the lab and looked at what actually comes out of aiokafka 0.14 while the container is paused. The first publish of a topic the producer has never seen ends in UnknownTopicOrPartitionError after request_timeout_ms, because _wait_on_metadata swallows the failed metadata refresh and only reports "no such topic"; once metadata is cached it is RequestTimedOutError, then NodeNotReadyError; a stopped broker gives NodeNotReadyError. None of those subclass the builtin ConnectionError or TimeoutError, so ErrorClassifier calls all of them permanent and the max_infra_retries loop in KafkaEventPublisher never runs for aiokafka's own errors — which is why your cycles were exactly 20 × 2 s. Second thing: publish_timeout defaults to 30 s and aiokafka's request_timeout_ms to 40 s, so with defaults the step's own timeout fires before the broker error ever surfaces, and that path counts too.

What I intend to do. A TransientError in omni_box.core.exceptions, the exception-shaped twin of handler_retry(count_as_attempt=False): a publisher or a handler raises it to say the failure is the environment's, not the event's. KafkaEventPublisher raises it once its retries are spent, for the connection, node-not-ready, request-timeout and client-timeout classes, and for an unknown topic only when a forced metadata refresh fails as well — a topic the broker answers about and still does not know keeps counting, so a typo still ends in failed. HandlerExecutionStep turns it into a non-counted failure with next_retry_at set. A PublisherExecutionStep subclass, which create_outbox_processor now installs, goes two steps further because everything in an outbox batch goes to the same broker: the publish timeout is transient too (a payload the broker cannot ack in 30 s is not the row's fault), and the first transient failure ends the cycle's publishing — the remaining rows are marked non-counted with the same error and their schedule untouched, so a dead broker costs one probe per cycle instead of a batch of them, and the rows are back the moment it answers.

On the backoff: the probed row goes back one second, nothing grows. The throttle on a dead broker is the probe itself, publish_timeout or the adapter's retry window, once per cycle; any row-level delay beyond a tick is pure lag after the broker returns. Growth would also need state that outlives a cycle, and in the documented wiring only the broker adapter does; I would rather not have an adapter setting the pipeline's schedule. If you want it later, a retry_after on the exception is where it would go.

Rejected: a classifier hook on the step, or the step consulting ErrorClassifier (core cannot see aiokafka's classes, and it would silently change how an inbox handler raising ConnectionError is counted); making publish return a handler result (changes the publisher contract for every adapter); treating UnknownTopicOrPartitionError as transient unconditionally (a missing topic would never fail and would cost a probe every cycle); rescheduling the whole batch along with the probed row (a misclassified row would drag every row behind it along).

Docs: rule 13 and the state paragraph on the agents page, the new step and exception in its tables, the retries section of troubleshooting with the operator story (an outage leaves rows pending with attempts_made unchanged; requeue_failed is for poisoned rows), the outbox section of the user guide, and a note on the two timeouts in configuration. A fix:, since no knob is added.

[AlexeyShalaev]

Shipped in #33, as planned. TransientError in omni_box.core.exceptions (re-exported at the top level) is the exception-shaped twin of handler_retry(count_as_attempt=False): a publisher or a handler raises it to say the failure is the environment's, not the event's, and the pipeline records it without spending an attempt. KafkaEventPublisher raises it once max_infra_retries are spent on a broker that does not answer — connection and node errors, request and client timeouts, and an unknown topic only when a forced metadata refresh fails as well, so a typo in a topic name still ends in failed. PublisherExecutionStep, which create_outbox_processor now installs, also treats the publish timeout as the broker's problem and ends the cycle's publishing on the first transient failure: everything left in the batch is recorded the same way, unsent, with its schedule untouched.

Your kafka_down_lab.py, unmodified, against the branch:

--- Kafka paused; the relay keeps ticking every cycle ---
  cycle 1 at   5.0 s: {'pending/attempts=0': 20}
  cycle 2 at  10.1 s: {'pending/attempts=0': 20}
  cycle 3 at  15.1 s: {'pending/attempts=0': 20}
  cycle 4 at  20.1 s: {'pending/attempts=0': 20}
  cycle 5 at  25.1 s: {'pending/attempts=0': 20}
  cycle 6 at  30.1 s: {'pending/attempts=0': 20}
  cycle 7 at  35.2 s: {'pending/attempts=0': 20}

--- Kafka is back ---
  next relay cycle: {'completed/attempts=0': 20}; messages on the topic: 20
  after requeue_failed on every failed row and one more cycle: {'completed/attempts=0': 20}; messages on the topic: 20

There is nothing left for requeue_failed to do — it goes back to being the operator's tool for genuinely poisoned rows. The outage is also 35 s of wall clock instead of 240 s, because a cycle is one probe now rather than twenty.

tests/integration/postgres/test_broker_outage.py holds your scenario against the real table; on master it fails exactly the way your table reads, {(PENDING, 1)} after the first cycle and {(FAILED, 6)} at the end. A broker that answers and rejects a record still spends an attempt per row — the third test in that file pins it, because that half of the behaviour was correct.

Nothing on your side changes, and no signature or default moved. One thing worth knowing while you wait for the release: publish_timeout defaults to 30 s and aiokafka's request_timeout_ms to 40 s, so with both defaults the step's timeout fires before the broker error surfaces. Both paths are now non-counted, so the budget is safe either way, but you get a more useful last_error if you set request_timeout_ms below publish_timeout. That is written up in the configuration guide, along with the outage story in troubleshooting and rule 13 on the agents page.