Repository navigation
feat(kafka): wire session timeout to all consumers [OMN-5445] - #905
Merged
jonahgabriel merged 4 commits intoMar 20, 2026
Merged
Conversation
… consumers [OMN-5445] Eliminate implicit reliance on aiokafka's aggressive 10s default session timeout across all AIOKafkaConsumer instances in the platform. The new platform defaults (30s session / 10s heartbeat) prevent rebalance storms caused by UnknownMemberIdError during transient processing delays. Three-surface fix: - ModelKafkaEventBusConfig: new fields with env var overrides (KAFKA_SESSION_TIMEOUT_MS, KAFKA_HEARTBEAT_INTERVAL_MS) and advisory heartbeat/session ratio validator - EventBusKafka: wire session/heartbeat/reconnect kwargs into consumer constructor; wire reconnect backoff kwargs into both producer constructors (previously configured but never passed) - 6 standalone consumers: add fields to BaseSettings configs and wire into AIOKafkaConsumer constructors - 3 ephemeral consumers: hardcode 30000/10000 defaults (no config surface) Includes 50 new unit tests covering config defaults, env overrides, field constraints, advisory validator, and constructor kwarg wiring.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
✅ Files skipped from review due to trivial changes (1)
📝 WalkthroughWalkthroughAdds Kafka session/heartbeat and reconnect tuning: new Pydantic fields and env overrides, propagation of session/heartbeat and reconnect_backoff kwargs into AIOKafka producer/consumer constructors, addition/export of a DLQ topic suffix, validation exemption, and tests for the new wiring. Changes
Sequence Diagram(s)(Skipped — changes are configuration wiring and small constructor argument additions; no new multi-component control flow requiring a sequence diagram.) Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
📝 Coding Plan
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (2)
tests/unit/event_bus/test_kafka_timeout_kwargs.py (1)
37-151: Consider adding cleanup to prevent resource leaks in tests.The tests create
EventBusKafkainstances and callbus.start()but don't callbus.close()afterward. While the mocks prevent actual Kafka connections, theEventBusKafkaclass may create asyncio tasks or other resources that should be cleaned up.♻️ Suggested: Add cleanup with try/finally or fixture
Option 1 — Add cleanup in each test:
async def test_consumer_receives_session_timeout_ms( self, bus: EventBusKafka, kafka_config: ModelKafkaEventBusConfig ) -> None: """Consumer constructor must receive session_timeout_ms from config.""" mock_consumer = MagicMock() mock_consumer.start = AsyncMock() - with patch( - "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" - ) as mock_producer_cls: - mock_producer = MagicMock() - mock_producer.start = AsyncMock() - mock_producer.stop = AsyncMock() - mock_producer_cls.return_value = mock_producer - await bus.start() + try: + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer" + ) as mock_producer_cls: + mock_producer = MagicMock() + mock_producer.start = AsyncMock() + mock_producer.stop = AsyncMock() + mock_producer_cls.return_value = mock_producer + await bus.start() + + with patch( + "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaConsumer", + return_value=mock_consumer, + ) as mock_consumer_cls: + await bus.subscribe( + "test-topic", on_message=AsyncMock(), group_id="test-group" + ) + call_kwargs = mock_consumer_cls.call_args + assert call_kwargs.kwargs["session_timeout_ms"] == 45000 + finally: + await bus.close()Option 2 — Update the fixture to yield and cleanup:
`@pytest.fixture` async def bus(kafka_config: ModelKafkaEventBusConfig) -> AsyncIterator[EventBusKafka]: """Create EventBusKafka instance with test config.""" bus = EventBusKafka(config=kafka_config) yield bus await bus.close()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/unit/event_bus/test_kafka_timeout_kwargs.py` around lines 37 - 151, Tests start EventBusKafka instances via bus.start() but never call bus.close(), risking leaked asyncio tasks; ensure each test cleans up by calling await bus.close() in a finally block or convert the bus fixture to a yield-style async fixture that yields EventBusKafka and calls await bus.close() after yield (reference: EventBusKafka, bus.start, bus.close, and the bus fixture).src/omnibase_infra/services/session/config_consumer.py (1)
62-81: Consider adding advisory validation for heartbeat/session ratio.The defaults are correctly configured (30000ms session timeout with 10000ms heartbeat = exactly the recommended 1:3 ratio). However, unlike
ModelKafkaEventBusConfigwhich has an advisory validator, this config allows users to set incompatible values without warning.For consistency with the event bus config, consider adding a
@model_validator(mode="after")that logs a warning whenheartbeat_interval_ms > session_timeout_ms / 3.♻️ Optional: Add advisory validator for ratio
`@model_validator`(mode="after") def validate_timing_relationships(self) -> Self: """Validate timing relationships between configuration values. ... """ batch_timeout_seconds = self.batch_timeout_ms / 1000 min_recommended_circuit_timeout = batch_timeout_seconds * 2 if self.circuit_breaker_timeout_seconds < min_recommended_circuit_timeout: logger.warning( "Circuit breaker timeout (%ds) is less than 2x batch timeout (%.1fs). " ... ) + + # Kafka recommends heartbeat_interval_ms <= session_timeout_ms / 3 + if self.heartbeat_interval_ms > self.session_timeout_ms / 3: + logger.warning( + "heartbeat_interval_ms (%d) exceeds Kafka's recommended maximum " + "of session_timeout_ms / 3 (%d). This may cause premature rebalances.", + self.heartbeat_interval_ms, + self.session_timeout_ms // 3, + ) return self🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/omnibase_infra/services/session/config_consumer.py` around lines 62 - 81, Add an advisory post-validation to the Pydantic model that contains session_timeout_ms and heartbeat_interval_ms: implement a `@model_validator`(mode="after") (same pattern as ModelKafkaEventBusConfig) on that config class which checks if heartbeat_interval_ms > session_timeout_ms / 3 and logs a warning (use the module/class logger) rather than raising an exception; reference the fields session_timeout_ms and heartbeat_interval_ms in the check and include their values in the warning message so users receive a clear advisory when their ratio is incompatible.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@src/omnibase_infra/services/session/config_consumer.py`:
- Around line 62-81: Add an advisory post-validation to the Pydantic model that
contains session_timeout_ms and heartbeat_interval_ms: implement a
`@model_validator`(mode="after") (same pattern as ModelKafkaEventBusConfig) on
that config class which checks if heartbeat_interval_ms > session_timeout_ms / 3
and logs a warning (use the module/class logger) rather than raising an
exception; reference the fields session_timeout_ms and heartbeat_interval_ms in
the check and include their values in the warning message so users receive a
clear advisory when their ratio is incompatible.
In `@tests/unit/event_bus/test_kafka_timeout_kwargs.py`:
- Around line 37-151: Tests start EventBusKafka instances via bus.start() but
never call bus.close(), risking leaked asyncio tasks; ensure each test cleans up
by calling await bus.close() in a finally block or convert the bus fixture to a
yield-style async fixture that yields EventBusKafka and calls await bus.close()
after yield (reference: EventBusKafka, bus.start, bus.close, and the bus
fixture).
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: be289d71-a7a0-4d8d-84d0-ad604761c0c0
📒 Files selected for processing (21)
src/omnibase_infra/event_bus/event_bus_kafka.pysrc/omnibase_infra/event_bus/models/config/model_kafka_event_bus_config.pysrc/omnibase_infra/projectors/snapshot_publisher_registration.pysrc/omnibase_infra/runtime/request_response_wiring.pysrc/omnibase_infra/services/observability/agent_actions/config.pysrc/omnibase_infra/services/observability/agent_actions/consumer.pysrc/omnibase_infra/services/observability/context_audit/config.pysrc/omnibase_infra/services/observability/context_audit/consumer.pysrc/omnibase_infra/services/observability/injection_effectiveness/config.pysrc/omnibase_infra/services/observability/injection_effectiveness/consumer.pysrc/omnibase_infra/services/observability/llm_cost_aggregation/config.pysrc/omnibase_infra/services/observability/llm_cost_aggregation/consumer.pysrc/omnibase_infra/services/observability/skill_lifecycle/config.pysrc/omnibase_infra/services/observability/skill_lifecycle/consumer.pysrc/omnibase_infra/services/session/config_consumer.pysrc/omnibase_infra/services/session/consumer.pysrc/omnibase_infra/tui/consumers/consumer_status.pysrc/omnibase_infra/validation/validation_exemptions.yamltests/unit/event_bus/test_kafka_config_session_timeout.pytests/unit/event_bus/test_kafka_timeout_kwargs.pytests/unit/services/test_consumer_session_timeout.py
…MN-5445] Moves two raw topic literal strings in observability consumer config files to the canonical constants in platform_topic_suffixes.py, fixing the Arch Invariants (OMN-3343) CI check that blocked this PR. - agent_actions/config.py: import and use SUFFIX_OMNICLAUDE_AGENT_ACTIONS_DLQ - skill_lifecycle/config.py: import and use new SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ - platform_topic_suffixes.py: add SUFFIX_OMNICLAUDE_SKILL_LIFECYCLE_DLQ constant and include it in _OMNICLAUDE_OBSERVABILITY_DLQ_TOPIC_SUFFIXES for provisioning - topic_literal_baseline.txt: update grandfathered line numbers after import insertion; remove now-fixed DLQ entries (lines 131/134 agent_actions, 124/126 skill_lifecycle) - check_contract_topic_parity.py: add skill-lifecycle-dlq to legacy allowlist (mirrors pattern for agent-actions-dlq and agent-observability-dlq)
jonahgabriel
enabled auto-merge
March 19, 2026 14:16
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/omnibase_infra/topics/platform_topic_suffixes.py`:
- Around line 648-650: Update the docstring on
ConfigSkillLifecycleConsumer.dlq_topic to remove the phrase "hardcoded default"
and instead state that the default is supplied via the centralized shared suffix
constant from platform_topic_suffixes (refer to the shared suffix constant, e.g.
SKILL_LIFECYCLE_DLQ_SUFFIX), so the wording reflects centralized configuration
rather than a hardcoded value.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: f44b57e2-81d1-4409-bc1f-53c58222f7c8
📒 Files selected for processing (5)
scripts/check_contract_topic_parity.pyscripts/validation/topic_literal_baseline.txtsrc/omnibase_infra/services/observability/agent_actions/config.pysrc/omnibase_infra/services/observability/skill_lifecycle/config.pysrc/omnibase_infra/topics/platform_topic_suffixes.py
✅ Files skipped from review due to trivial changes (2)
- scripts/check_contract_topic_parity.py
- scripts/validation/topic_literal_baseline.txt
🚧 Files skipped from review as they are similar to previous changes (2)
- src/omnibase_infra/services/observability/agent_actions/config.py
- src/omnibase_infra/services/observability/skill_lifecycle/config.py
… __init__ [OMN-5445]
…hitelist [OMN-5445]
jonahgabriel
added this pull request to the merge queue
Mar 19, 2026
github-merge-queue
Bot
removed this pull request from the merge queue due to no response for status checks
Mar 19, 2026
jonahgabriel
added this pull request to the merge queue
Mar 19, 2026
jonahgabriel
deleted the
jonahgabriel/omn-5445-kafka-consumer-session-timeout-fix
branch
March 20, 2026 00:50
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
1 participant
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.
Summary
session_timeout_ms(30s) andheartbeat_interval_ms(10s) toModelKafkaEventBusConfigwith env var overrides (KAFKA_SESSION_TIMEOUT_MS,KAFKA_HEARTBEAT_INTERVAL_MS) and advisory heartbeat/session ratio validatorsession_timeout_ms,heartbeat_interval_ms,reconnect_backoff_ms,reconnect_backoff_max_msintoEventBusKafkaconsumer constructor; wirereconnect_backoff_ms/reconnect_backoff_max_msinto both producer constructors (previously configured but never passed)BaseSettingsconfigs and wire into theirAIOKafkaConsumerconstructorsModelKafkaEventBusConfigmethod count (pre-existing borderline at 10 methods)Test plan
ModelKafkaEventBusConfigsession timeout (defaults, constraints, env overrides, advisory validator, JSON schema)EventBusKafkaconsumer/producer kwargs wiring (session_timeout, heartbeat, reconnect backoff)Summary by CodeRabbit
New Features
Validation
Tests