Skip to content

fix(OMN-15232): fail-close the deserialization-path DLQ publish in event_bus_kafka #2497

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jonahgabriel merged 1 commit into from
Jul 27, 2026
Merged

fix(OMN-15232): fail-close the deserialization-path DLQ publish in event_bus_kafka #2497

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
197 changes: 192 additions & 5 deletions src/omnibase_infra/event_bus/event_bus_kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,6 +206,7 @@ async def handler(msg):
KafkaError,
UnknownTopicOrPartitionError,
)
from aiokafka.structs import TopicPartition

from omnibase_infra.enums import EnumConsumerGroupPurpose, EnumInfraTransportType
from omnibase_infra.errors import (
Expand Down Expand Up @@ -253,6 +254,11 @@ async def handler(msg):

logger = logging.getLogger(__name__)

# OMN-15232: pause between fail-closed rewinds so a persistently unreachable DLQ
# does not turn the consume loop into a hot spin over the same offset. The
# rewind itself is what keeps the data safe; this only bounds the retry rate.
DLQ_UNPERSISTED_REWIND_BACKOFF_SECONDS: float = 1.0


class EventBusKafka(
MixinKafkaBroadcast,
Expand Down Expand Up @@ -2047,8 +2053,19 @@ async def _dispatch_to_subscriber(
topic: str,
group_id: str,
correlation_id: UUID,
) -> None:
"""Invoke a single subscriber callback, routing to DLQ on exhausted retries."""
) -> bool:
"""Invoke a single subscriber callback, routing to DLQ on exhausted retries.

Returns:
``True`` when it is safe for the partition offset to advance past
this message — the callback succeeded, retries remain (so the
message is still live), or the DLQ write for an exhausted message
was confirmed durable. ``False`` when retries were exhausted AND
the DLQ write was NOT confirmed: the message then exists nowhere
durable, so the caller must rewind rather than let the offset move
(OMN-15232, same discipline as the OMN-14936 gate in
``runtime/event_bus_subcontract_wiring.py``).
"""
try:
await callback(event_message)
except Exception as e:
Expand All @@ -2072,13 +2089,42 @@ async def _dispatch_to_subscriber(
)

if retries_exhausted:
await self._publish_to_dlq(
dlq_result = await self._publish_to_dlq(
original_topic=topic,
failed_message=event_message,
error=e,
correlation_id=correlation_id,
consumer_group=group_id,
)
# OMN-15232: only an explicit ``False`` counts as a confirmed
# non-persist -- duck-typed hosts/test doubles that still
# return ``None`` keep their prior behavior, matching the
# allowance the OMN-14936 gate makes in
# runtime/event_bus_subcontract_wiring.py.
dlq_persisted = dlq_result is not False
if not dlq_persisted:
# TRY400 suppressed below: the handler traceback was
# already emitted by the logger.exception above; this line
# reports the DLQ persistence outcome, not a second copy of
# that stack.
logger.error( # noqa: TRY400
"dlq_publish_not_persisted topic=%s subscription_id=%s "
"correlation_id=%s error_type=%s -- retries exhausted and "
"the DLQ write was NOT confirmed; offset will be rewound "
"instead of advancing (OMN-15232)",
topic,
subscription_id,
str(correlation_id),
type(e).__name__,
extra={
"topic": topic,
"group_id": group_id,
"subscription_id": subscription_id,
"correlation_id": str(correlation_id),
"error_type": type(e).__name__,
},
)
return dlq_persisted
else:
logger.warning(
f"Handler failed but retries available ({retry_count}/{max_retries})",
Expand All @@ -2090,6 +2136,110 @@ async def _dispatch_to_subscriber(
},
)

return True

async def _rewind_after_unpersisted_dlq(
self,
consumer: AIOKafkaConsumer,
msg: object,
topic: str,
group_id: str,
correlation_id: UUID,
failure_stage: str,
) -> None:
"""Withhold offset advancement after an unconfirmed DLQ write (OMN-15232).

The consumers this class builds run with
``enable_auto_commit=self._config.enable_auto_commit``, which defaults
to ``True``: the client commits the *fetch position* on its own cadence,
and this loop never calls ``commit`` itself. So — unlike the manual-commit
path in ``runtime/event_bus_subcontract_wiring.py``, where the OMN-14936
gate simply returns without committing — merely declining to commit here
does nothing. The offset advances anyway and the message is lost.

The fail-closed action that works under BOTH commit models is a rewind of
the fetch position to the failed message's own offset
(``consumer.seek(tp, msg.offset)``) — the identical "does NOT advance the
committed offset" mechanism ``KafkaTransport.nack`` uses in
``event_bus/kafka_transport.py``. Under auto-commit the committer then
commits the rewound position, which cannot be past the message; under
manual commit the message is simply refetched. Either way Kafka
redelivers and the DLQ write is retried instead of the record being
silently dropped.

Only the failed message's own partition is rewound; sibling partitions
are untouched (same per-partition discipline as OMN-14757).
"""
msg_topic = getattr(msg, "topic", None) or topic
partition = getattr(msg, "partition", None)
offset = getattr(msg, "offset", None)

if partition is None or offset is None:
logger.error(
"dlq_unpersisted_rewind_impossible topic=%s stage=%s "
"correlation_id=%s -- message carries no partition/offset "
"coordinate, so offset advancement cannot be withheld; this "
"record may be lost (OMN-15232)",
topic,
failure_stage,
str(correlation_id),
extra={
"topic": topic,
"group_id": group_id,
"correlation_id": str(correlation_id),
"failure_stage": failure_stage,
},
)
return

try:
consumer.seek(TopicPartition(msg_topic, int(partition)), int(offset))
except Exception as seek_error:
logger.exception(
"dlq_unpersisted_rewind_failed topic=%s partition=%s offset=%s "
"stage=%s correlation_id=%s error=%s -- could NOT withhold offset "
"advancement after an unconfirmed DLQ write; this record may be "
"lost (OMN-15232)",
msg_topic,
partition,
offset,
failure_stage,
str(correlation_id),
str(seek_error),
extra={
"topic": msg_topic,
"group_id": group_id,
"partition": partition,
"offset": offset,
"failure_stage": failure_stage,
"correlation_id": str(correlation_id),
},
)
return

logger.error(
"dlq_unpersisted_offset_rewound topic=%s partition=%s offset=%s "
"stage=%s correlation_id=%s -- DLQ persistence was NOT confirmed; "
"fetch position rewound so the offset cannot advance past an "
"undelivered DLQ write (OMN-15232)",
msg_topic,
partition,
offset,
failure_stage,
str(correlation_id),
extra={
"topic": msg_topic,
"group_id": group_id,
"partition": partition,
"offset": offset,
"failure_stage": failure_stage,
"correlation_id": str(correlation_id),
},
)

if DLQ_UNPERSISTED_REWIND_BACKOFF_SECONDS > 0:
await asyncio.sleep(DLQ_UNPERSISTED_REWIND_BACKOFF_SECONDS)

async def _emit_consume_loop_error_health_event(
self,
topic: str,
Expand Down Expand Up @@ -2230,26 +2380,63 @@ async def _consume_loop(
)
# Deserialization errors are permanent failures - route to DLQ
# Create minimal message from raw Kafka data for DLQ context
await self._publish_raw_to_dlq(
dlq_result = await self._publish_raw_to_dlq(
original_topic=topic,
raw_msg=msg,
error=e,
correlation_id=correlation_id,
failure_type="deserialization_error",
consumer_group=effective_consumer_group,
)
# OMN-15232: an undeserializable message that did NOT reach
# the DLQ exists nowhere durable. Skipping it here while the
# client auto-commits the position is a silent, committed,
# unrecoverable drop -- the OMN-14936 failure mode at a call
# site that fix did not cover. Rewind instead so Kafka
# redelivers and the DLQ write is retried. Only an explicit
# ``False`` counts as a confirmed non-persist (duck-typed
# hosts returning ``None`` keep prior behavior).
if dlq_result is False:
await self._rewind_after_unpersisted_dlq(
consumer,
msg,
topic,
group_id,
correlation_id,
"deserialization_error",
)
continue # Skip this message but continue consuming

# Dispatch to all subscribers
offset_may_advance = True
for _sub_group_id, subscription_id, callback in subscribers:
await self._dispatch_to_subscriber(
dispatch_offset_safe = await self._dispatch_to_subscriber(
callback,
subscription_id,
event_message,
topic,
group_id,
correlation_id,
)
# OMN-15232: same gate on the dispatch path. Every subscriber
# still gets the message (one failing subscriber must not
# starve the others), but if ANY of them exhausted retries
# without a confirmed DLQ record, the offset must not move.
# Redelivery may duplicate for the subscribers that
# succeeded -- at-least-once, which is the delivery contract
# here, and strictly preferable to losing the record.
if dispatch_offset_safe is False:
offset_may_advance = False

if not offset_may_advance:
await self._rewind_after_unpersisted_dlq(
consumer,
msg,
topic,
group_id,
correlation_id,
"handler_retries_exhausted",
)

except asyncio.CancelledError:
# Graceful cancellation - this is expected during shutdown
Expand Down
24 changes: 21 additions & 3 deletions src/omnibase_infra/event_bus/mixin_kafka_dlq.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,7 @@ async def _publish_to_dlq(
dlq_topic: str | None = None,
failure_class: str | None = None,
validation_detail: str | None = None,
) -> None:
) -> bool:
"""Publish failed message to dead letter queue with metrics and alerting.

This method publishes messages that failed processing to the configured
Expand Down Expand Up @@ -261,9 +261,21 @@ async def _publish_to_dlq(
validation_detail: Optional real pydantic ``ValidationError``
detail (OMN-14492) for the ``publisher_malformed`` case.

Returns:
``True`` if the DLQ message was actually published (producer send
acked within timeout, either on the resolved topic or the category
fallback), ``False`` otherwise (rejected input, producer
unavailable, or every send failed/timed out). Mirrors the
``_publish_raw_to_dlq`` contract added under OMN-14936: callers
that gate offset advancement on durable DLQ persistence
(OMN-15232) MUST check this return value instead of assuming the
write succeeded just because no exception escaped this method.

Note:
This method logs errors if DLQ publishing fails but does not raise
exceptions to prevent cascading failures in the consumer loop.
exceptions to prevent cascading failures in the consumer loop. The
boolean return value is the only signal of actual persistence
success.
"""
# Validate original_topic - reject whitespace-only values
if not original_topic or not original_topic.strip():
Expand All @@ -274,7 +286,7 @@ async def _publish_to_dlq(
"error_type": type(error).__name__,
},
)
return
return False

# Track timing for metrics
start_time = datetime.now(UTC)
Expand Down Expand Up @@ -625,6 +637,12 @@ async def _publish_to_dlq(
# Invoke DLQ callbacks for custom alerting
await self._invoke_dlq_callbacks(dlq_event)

# OMN-15232: surface the real persistence outcome so callers can gate
# offset advancement on it. ``success`` is True only when a
# send_and_wait ack came back within the timeout, on either the
# resolved topic or the category fallback.
return success

async def _invoke_dlq_callbacks(self, event: ModelDlqEvent) -> None:
"""Invoke registered DLQ callbacks with error isolation.

Expand Down
Loading
Loading