event_queue: consumers never recover when a kafka topic is deleted/recreated

[extreme4all]

Context

While testing the new firehose websocket service against the local docker-compose stack, two consumer-resilience gaps surfaced in components/bot_detector/event_queue. One is fixed; the topic-lifecycle one is still open.

1. Poison messages stall consumers forever (FIXED)

AIOKafkaConsumerAdapter used value_deserializer=lambda x: orjson.loads(x). A single malformed message raises inside aiokafka's internal fetcher task, which dies silently — the consumer then blocks in get_one() forever with no error surfaced to callers.

Fixed in components/bot_detector/event_queue/adapters/kafka/adapter.py: _deserialize_value() never raises and returns a {"_poison": true, "raw": ...} envelope that fails model validation downstream, so callers receive a normal error-as-value (ValidationError / ConsumerFetchError path) instead of a dead consumer.

2. Topic deleted/recreated under a live consumer (OPEN)

Reproduction (dev compose):

• kafka_setup runs with RESET_TOPICS=true on every docker compose up, deleting and recreating all topics (including players.scraped) while running consumers are subscribed.
• After recreation, consumers spam forever and never heal without a restart:

{"lvl": "ERROR", "name": "aiokafka.cluster", "msg": "Topic players.scraped not found in cluster metadata"}
{"lvl": "ERROR", "name": "aiokafka.consumer.group_coordinator", "func": "_do_commit_offsets",
 "msg": "OffsetCommit failed for group firehose-anonymous on partition ... : UnknownTopicOrPartitionError"}

Problems in event_queue today:

• AIOKafkaConsumerAdapter.get_one() wraps every failure into a generic ConsumerFetchError, so callers cannot distinguish "transient fetch error" from "the topic I'm subscribed to no longer exists".
• Commit errors are logged by aiokafka but the adapter keeps committing to stale topic ids forever; there is no re-subscribe / re-join / recovery strategy.

Proposal

1. Surface typed errors (or error metadata) from the adapter for UnknownTopicOrPartitionError / InvalidTopicError / UnknownTopicError so callers can decide: skip, backoff, or rebuild the consumer.
2. Add a recovery strategy in the adapter (e.g. on topic-gone errors: pause, wait for the topic to reappear in metadata, re-subscribe) or document explicitly that callers must rebuild the consumer.
3. Add tests mocking aiokafka raising UnknownTopicOrPartitionError from getone() / commit().

Workaround used in projects/firehose for now: shared per-group consumers with skip+backoff loops and a fixed shared stream (firehose-anonymous), plus RESET_TOPICS=false in compose.