Repository navigation
fix(OMN-11319): track Pattern B terminal topics - #1682
Merged
Merged
Conversation
📝 WalkthroughWalkthroughRefreshes Kafka terminal topic metadata via a new helper that prefers consumer._client.set_topics/add_topic (fallback to consumer.topics()), updates RuntimePatternBBroker to call it during retries, and adjusts unit/integration test fakes and assertions to track and assert set_topics usage. Adds a contract file OMN-11319. ChangesPattern B terminal topic tracking fix
Sequence Diagram(s)sequenceDiagram
participant AssignPartitions as _assign_terminal_topic_partitions
participant RefreshHelper as _refresh_terminal_topic_metadata
participant ConsumerClient as consumer._client
participant ConsumerType as consumer.topics()
AssignPartitions->>RefreshHelper: refresh_terminal_topic_metadata(consumer, topics)
RefreshHelper->>ConsumerClient: check for set_topics
alt set_topics exists
RefreshHelper->>ConsumerClient: set_topics(topics)
ConsumerClient-->>RefreshHelper: awaitable/result
else check add_topic
RefreshHelper->>ConsumerClient: check add_topic
RefreshHelper->>ConsumerClient: add_topic(topics)
ConsumerClient-->>RefreshHelper: awaitable/result
else fallback
RefreshHelper->>ConsumerType: topics()
ConsumerType-->>RefreshHelper: list/awaitable
end
RefreshHelper-->>AssignPartitions: metadata refreshed
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
jonahgabriel
enabled auto-merge (squash)
May 21, 2026 01:53
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
♻️ Duplicate comments (1)
src/omnibase_infra/runtime/service_pattern_b_broker.py (1)
334-343:⚠️ Potential issue | 🟠 Major | ⚡ Quick winKeep awaitable topic-refresh hooks supported.
If a consumer client exposes
async set_topics()orasync add_topic(), these calls now create a coroutine and return immediately, so metadata never refreshes and the waiter can spin until timeout. Restoring the old “await if awaitable” behavior keeps the sync AIOKafka path while preserving compatibility with async fakes/transports.Suggested fix
set_topics = getattr(client, "set_topics", None) if callable(set_topics): - set_topics(list(topics)) + result = set_topics(list(topics)) + if isinstance(result, Awaitable): + await result return add_topic = getattr(client, "add_topic", None) if callable(add_topic): for topic in topics: - add_topic(topic) + result = add_topic(topic) + if isinstance(result, Awaitable): + await result return#!/bin/bash set -euo pipefail printf '\n== async set_topics/add_topic implementations ==\n' rg -n -C2 --type=py 'async def (set_topics|add_topic)\(' || true printf '\n== sync set_topics/add_topic implementations ==\n' rg -n -C2 --type=py 'def (set_topics|add_topic)\(' || true printf '\n== metadata refresh helper and nearby call sites ==\n' rg -n -C2 --type=py '_refresh_terminal_topic_metadata|set_topics\(|add_topic\(' || trueIf the first section shows any reachable async implementation, the current helper is dropping that refresh path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/omnibase_infra/runtime/service_pattern_b_broker.py` around lines 334 - 343, The current helper calls client.set_topics(...) and client.add_topic(...) but drops the coroutine if those methods are async; change the calls to capture the return value and await it when awaitable so async implementations are supported: for set_topics call result = set_topics(list(topics)) and if inspect.isawaitable(result) (or asyncio.iscoroutine(result)) await result; for add_topic call result = add_topic(topic) inside the loop and await each result when awaitable (or collect and await them sequentially). Ensure you import inspect or asyncio and preserve the existing sync behavior in the branches that use set_topics, add_topic, and the surrounding metadata refresh/_refresh_terminal_topic_metadata logic.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Duplicate comments:
In `@src/omnibase_infra/runtime/service_pattern_b_broker.py`:
- Around line 334-343: The current helper calls client.set_topics(...) and
client.add_topic(...) but drops the coroutine if those methods are async; change
the calls to capture the return value and await it when awaitable so async
implementations are supported: for set_topics call result =
set_topics(list(topics)) and if inspect.isawaitable(result) (or
asyncio.iscoroutine(result)) await result; for add_topic call result =
add_topic(topic) inside the loop and await each result when awaitable (or
collect and await them sequentially). Ensure you import inspect or asyncio and
preserve the existing sync behavior in the branches that use set_topics,
add_topic, and the surrounding metadata refresh/_refresh_terminal_topic_metadata
logic.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5f79cfca-2922-47f3-b7f4-21a0ce7f465a
📒 Files selected for processing (3)
src/omnibase_infra/runtime/service_pattern_b_broker.pytests/integration/test_pattern_b_broker_terminal_waiter.pytests/unit/runtime/test_service_pattern_b_broker.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Evidence-Source: OCC#1245
Evidence-Ticket: OMN-11319
Summary
Verification
OCC Evidence
Runtime Evidence
Post-merge deploy proof still required: rebuild/restart the .201 stability runtime from this merged change, rerun /skill smoke, and verify the worker command plus terminal event for the smoke correlation.