{"record":{"id":"de09135d5f4fe0fd","repo":"xai-org/x-algorithm","slug":"failed-to-create-consumer-for-thread-de0913","errorCode":null,"errorMessage":"Failed to create consumer for thread {}: {:#}","messagePattern":"Failed to create consumer for thread (.+?): (.+?)","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"critical","filePath":"thunder/kafka/tweet_events_listener_v2.rs","lineNumber":103,"sourceCode":"                    );\n\n                    if let Err(e) = process_tweet_events_v2(\n                        consumer,\n                        post_store_clone,\n                        batch_size,\n                        tx_clone,\n                        semaphore_clone,\n                    )\n                    .await\n                    {\n                        panic!(\n                            \"Tweet events processing thread {} exited unexpectedly: {:#}. This is a critical failure - the feeder cannot function without tweet event processing.\",\n                            thread_id, e\n                        );\n                    }\n                }\n                Err(e) => {\n                    panic!(\n                        \"Failed to create consumer for thread {}: {:#}\",\n                        thread_id, e\n                    );\n                }\n            }\n        });\n    }\n}\n\nfn deserialize_batch(\n    messages: Vec<KafkaMessage>,\n) -> Result<(Vec<LightPost>, Vec<TweetDeleteEvent>)> {\n    let start_time = Instant::now();\n    let num_messages = messages.len();\n    let results = deserialize_kafka_messages(messages, deserialize_tweet_event_v2)?;\n    let deser_elapsed = start_time.elapsed();\n    if DESER_LOG_COUNTER\n        .fetch_add(1, Ordering::Relaxed)","sourceCodeStart":85,"sourceCodeEnd":121,"githubUrl":"https://github.com/xai-org/x-algorithm/blob/24c60942c5c5fdad3a6addffb4c6e6d2f228f04f/thunder/kafka/tweet_events_listener_v2.rs#L85-L121","documentation":"This panic occurs when one of the tweet-event processing worker threads fails to create its rd_kafka consumer during startup of the v2 tweet events listener. Consumer creation in rdkafka can fail for reasons such as an invalid broker address, bad consumer group config, authentication failure, or DNS resolution problems. Because the feeder cannot function without tweet event processing, the code treats this as a fatal, non-recoverable condition and panics inside the spawned thread.","triggerScenarios":"Calling start_tweet_event_processing_v2 with a Kafka bootstrap broker that is unreachable/misconfigured, an invalid group.id or auth credentials, or a broker version incompatibility. Each spawned thread calls its consumer constructor; the Err branch of that constructor panics with 'Failed to create consumer for thread {thread_id}'.","commonSituations":"Misconfigured KAFKA_BOOTSTRAP_SERVERS env var, typos in broker hostnames, missing SASL/SSL config, Kafka cluster not yet up when the service starts, or a network/firewall block between the service and the broker.","solutions":["Verify the Kafka bootstrap servers address and connectivity (kcat -L -b $BROKER or telnet) from the host running the feeder.","Check the consumer config passed to the thread (group.id, SASL/SSL settings) against the cluster's requirements.","Ensure Kafka is reachable at startup; add retry/backoff around consumer creation or delay service start until the broker is available.","Inspect the {:#} formatted rdkafka error in the panic output — it names the exact cause (e.g. 'InvalidData', 'Broker not available').","If brokers flake at boot, consider catching the error in spawn and retrying thread creation instead of panicking."],"exampleFix":"// before\nErr(e) => {\n    panic!(\"Failed to create consumer for thread {}: {:#}\", thread_id, e);\n}\n\n// after (retry with backoff instead of aborting)\nErr(e) => {\n    error!(\"Consumer creation failed for thread {}: {:#}; retrying\", thread_id, e);\n    continue; // or implement bounded retry before panicking\n}","handlingStrategy":"validation","validationCode":"// Before starting processing, verify broker reachability\nuse std::net::TcpStream;\nfn broker_reachable(addr: &str) -> bool {\n    TcpStream::connect(addr).is_ok()\n}\nassert!(broker_reachable(kafka_broker_addr), \"Kafka broker unreachable\");\nstart_tweet_event_processing_v2(...);","typeGuard":null,"tryCatchPattern":"// Rust: catch_unwind around thread spawn to log and retry instead of crashing the process\nlet result = std::panic::catch_unwind(|| spawn_processing_threads_v2(cfg));\nif result.is_err() { /* re-init with backoff, alert on-call */ }","preventionTips":["Run preflight connectivity checks to the Kafka broker before spawning consumer threads.","Validate consumer config (group.id, SASL, SSL) in CI against a staging broker.","Container startup ordering: wait for Kafka readiness probe before starting the feeder.","Centralize broker/env config so typos fail validation, not runtime panics."],"tags":["kafka","rust","consumer","startup","panic","rdkafka"],"backgroundTag":"kafka-consumer-creation-failed","analyzedSha":"24c60942c5c5fdad3a6addffb4c6e6d2f228f04f","analyzedAt":"2026-08-28T11:40:14.686Z","schemaVersion":2},"datasetVersion":"2026-08-28T16:17:29.566Z"}