fix(OMN-12533): retry broker metadata startup errors

docker/Dockerfile.runtime

@@ -108,6 +108,8 @@ RUN set -eu; \
     fi
 
 # Build-time environment configuration
+# Keep uv's network timeout aligned with CI install steps for large public-index
+# wheel installs during runtime image builds.
 ENV PYTHONUNBUFFERED=1 \
     PYTHONDONTWRITEBYTECODE=1 \
     PIP_NO_CACHE_DIR=1 \

docker/docker-compose.infra.yml

@@ -153,9 +153,10 @@ x-runtime-env: &runtime-env
   ONEX_RUNTIME_CONTRACTS_DIR: /app/contracts/runtime
   ONEX_LOG_LEVEL: ${ONEX_LOG_LEVEL:-INFO}
   ONEX_ENVIRONMENT: ${ONEX_ENVIRONMENT:-local}
-  # Local Docker Redpanda can cap contract topic partitions on constrained laptops.
-  # Use 0 to keep contract-declared partition counts.
-  ONEX_TOPIC_PROVISIONER_MAX_PARTITIONS: ${ONEX_TOPIC_PROVISIONER_MAX_PARTITIONS:-0}
+  # Local Docker Redpanda uses a bounded single-shard broker. Cap default topic
+  # creation to one partition unless an operator explicitly opts into wider
+  # contract-declared partition counts.
+  ONEX_TOPIC_PROVISIONER_MAX_PARTITIONS: ${ONEX_TOPIC_PROVISIONER_MAX_PARTITIONS:-1}
   # Linear API for build loop ticket fetching
   LINEAR_API_KEY: ${LINEAR_API_KEY:?LINEAR_API_KEY must be set}
   # GitHub API/CLI auth for runtime-owned PR lifecycle inventory and merge gates.

src/omnibase_infra/event_bus/event_bus_kafka.py

@@ -202,6 +202,7 @@ async def handler(msg):
 from aiokafka import AIOKafkaConsumer, AIOKafkaProducer
 from aiokafka.abc import AbstractTokenProvider
 from aiokafka.errors import (
+    BrokerNotAvailableError,
     InvalidPartitionsError,
     KafkaError,
     UnknownTopicOrPartitionError,
@@ -1751,22 +1752,25 @@ async def _start_consumer_for_topic_unlocked(
             **self._build_auth_kwargs(),
         )
 
-        # Redpanda (and Kafka) can return UnknownTopicOrPartitionError or
-        # InvalidPartitionsError during consumer.start() when partition metadata
-        # has not yet propagated after topic creation. Both errors are transient
-        # in Redpanda --smp 1 --mode dev-container: UnknownTopicOrPartitionError
-        # appears when metadata hasn't propagated yet; InvalidPartitionsError
+        # Redpanda (and Kafka) can return UnknownTopicOrPartitionError,
+        # InvalidPartitionsError, or BrokerNotAvailableError during
+        # consumer.start() when topic or partition metadata has not yet
+        # propagated after topic creation. These are transient in Redpanda
+        # --smp 1 --mode dev-container: UnknownTopicOrPartitionError appears
+        # when metadata has not propagated yet; InvalidPartitionsError
         # (Kafka error 37) appears when the broker acknowledges the topic exists
-        # but hasn't finalized partition metadata. Neither error is permanent —
-        # both resolve once Redpanda stabilizes the topic state.
+        # but has not finalized partition metadata; BrokerNotAvailableError
+        # appears while the broker cannot yet serve metadata for the topic.
+        # None is permanent when topic provisioning is still converging.
         #
         # aiokafka raises these from _wait_topics() inside consumer.start(). They
         # propagate out of asyncio.wait_for before the timeout fires, so
         # KAFKA_TIMEOUT_SECONDS alone does not help.
         #
-        # Fix: retry both errors with exponential backoff within the timeout
-        # budget. Each failed attempt stops and discards the consumer; a fresh
-        # AIOKafkaConsumer is created for each retry to avoid stale state.
+        # Fix: retry metadata errors with exponential backoff within the
+        # timeout budget. Each failed attempt stops and discards the consumer;
+        # a fresh AIOKafkaConsumer is created for each retry to avoid stale
+        # state.
         metadata_retry_deadline = (
             asyncio.get_event_loop().time() + self._timeout_seconds
         )
@@ -1793,7 +1797,11 @@ async def _start_consumer_for_topic_unlocked(
                     )
                 raise
 
-            except (UnknownTopicOrPartitionError, InvalidPartitionsError) as e:
+            except (
+                UnknownTopicOrPartitionError,
+                InvalidPartitionsError,
+                BrokerNotAvailableError,
+            ) as e:
                 # Transient metadata propagation lag — stop the failed consumer,
                 # wait briefly, and retry if still within the timeout budget.
                 error_type = type(e).__name__

src/omnibase_infra/event_bus/service_topic_manager.py

@@ -414,9 +414,10 @@ async def ensure_topic_exists(
                     topic_configs=config.to_kafka_config(),
                 )
             else:
+                default_spec = ModelTopicSpec(suffix=topic_name)
                 new_topic = NewTopic(
                     name=topic_name,
-                    num_partitions=DEFAULT_EVENT_TOPIC_PARTITIONS,
+                    num_partitions=self._creation_partitions(default_spec),
                     replication_factor=DEFAULT_EVENT_TOPIC_REPLICATION_FACTOR,
                 )
 

tests/integration/event_bus/test_kafka_invalid_partitions_retry_integration.py

@@ -1,10 +1,10 @@
 # SPDX-FileCopyrightText: 2025 OmniNode.ai Inc.
 # SPDX-License-Identifier: MIT
-"""Integration test for InvalidPartitionsError retry in consumer start (OMN-9120).
+"""Integration tests for transient metadata retry in consumer start.
 
 Verifies that EventBusKafka._start_consumer_for_topic retries when the broker
-returns InvalidPartitionsError (Kafka error 37), which Redpanda --smp 1
---mode dev-container raises for topics auto-created with > 1 partition.
+returns transient metadata errors, which Redpanda --smp 1 --mode dev-container
+can raise while topic metadata converges.
 
 Without the fix, InvalidPartitionsError fell to the generic except Exception
 block and raised InfraConnectionError immediately. With the fix, it enters the
@@ -20,7 +20,7 @@
 from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
-from aiokafka.errors import InvalidPartitionsError
+from aiokafka.errors import BrokerNotAvailableError, InvalidPartitionsError
 
 from omnibase_infra.errors import InfraTimeoutError
 from omnibase_infra.event_bus.event_bus_kafka import EventBusKafka
@@ -98,6 +98,63 @@ def make_consumer(*args: object, **kwargs: object) -> MagicMock:
     consumer_second.start.assert_awaited_once()
 
 
+@pytest.mark.asyncio
+async def test_broker_not_available_error_is_retried_and_recovers(
+    kafka_config: ModelKafkaEventBusConfig,
+) -> None:
+    """Consumer start recovers after BrokerNotAvailableError on first attempt.
+
+    Redpanda can return BrokerNotAvailableError while an auto-created topic is
+    visible to the broker but metadata is not yet stable enough for aiokafka's
+    consumer startup path.
+    """
+    bus = EventBusKafka(config=kafka_config)
+
+    producer_mock = MagicMock()
+    producer_mock.start = AsyncMock()
+    producer_mock.stop = AsyncMock()
+
+    with patch(
+        "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaProducer",
+        return_value=producer_mock,
+    ):
+        await bus.start()
+
+    consumer_first = MagicMock()
+    consumer_first.start = AsyncMock(side_effect=BrokerNotAvailableError())
+    consumer_first.stop = AsyncMock()
+
+    consumer_second = MagicMock()
+    consumer_second.start = AsyncMock()
+    consumer_second.stop = AsyncMock()
+
+    call_count = 0
+
+    def make_consumer(*args: object, **kwargs: object) -> MagicMock:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            return consumer_first
+        return consumer_second
+
+    with (
+        patch(
+            "omnibase_infra.event_bus.event_bus_kafka.AIOKafkaConsumer",
+            side_effect=make_consumer,
+        ),
+        patch("asyncio.sleep", new_callable=AsyncMock),
+    ):
+        await bus.subscribe(
+            "onex.evt.integration.broker-not-available-retry.v1",
+            on_message=AsyncMock(),
+            group_id="integration-broker-not-available-retry-group",
+        )
+
+    assert call_count == 2
+    consumer_first.stop.assert_awaited_once()
+    consumer_second.start.assert_awaited_once()
+
+
 @pytest.mark.asyncio
 async def test_invalid_partitions_error_exhausts_budget_raises_infra_timeout(
     kafka_config: ModelKafkaEventBusConfig,

tests/unit/event_bus/test_service_topic_manager.py

@@ -338,11 +338,17 @@ async def test_ensure_provisioned_topics_already_exist(
         assert len(result["created"]) == 0
         assert len(result["failed"]) == 0
 
-    async def test_ensure_single_topic_success(self, contracts_root: Path) -> None:
+    async def test_ensure_single_topic_success(
+        self,
+        contracts_root: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
         """Single topic creation returns True on success."""
+        monkeypatch.setenv("ONEX_TOPIC_PROVISIONER_MAX_PARTITIONS", "1")
         manager = _make_provisioner(contracts_root)
 
         mock_admin_cls = MagicMock()
+        mock_new_topic_cls = MagicMock()
         mock_admin_instance = AsyncMock()
         mock_admin_instance.start = AsyncMock()
         mock_admin_instance.close = AsyncMock()
@@ -355,7 +361,7 @@ async def test_ensure_single_topic_success(self, contracts_root: Path) -> None:
                 "aiokafka": MagicMock(),
                 "aiokafka.admin": MagicMock(
                     AIOKafkaAdminClient=mock_admin_cls,
-                    NewTopic=MagicMock(),
+                    NewTopic=mock_new_topic_cls,
                 ),
                 "aiokafka.errors": MagicMock(
                     TopicAlreadyExistsError=type(
@@ -367,6 +373,11 @@ async def test_ensure_single_topic_success(self, contracts_root: Path) -> None:
             result = await manager.ensure_topic_exists("test.topic")
 
         assert result is True
+        mock_new_topic_cls.assert_called_once_with(
+            name="test.topic",
+            num_partitions=1,
+            replication_factor=1,
+        )
 
     async def test_ensure_single_topic_failure(self, contracts_root: Path) -> None:
         """Single topic creation returns False on failure."""