{"record":{"id":"5c2e4195bce20b1b","repo":"quickwit-oss/quickwit","slug":"consumer-was-dropped","errorCode":null,"errorMessage":"consumer was dropped","messagePattern":"consumer was dropped","errorType":"exception","errorClass":"ActorExitStatus","httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/kafka_source.rs","lineNumber":468,"sourceCode":"            .send_publish_lock(NewPublishLock(publish_lock), ctx)\n            .await?;\n        Ok(())\n    }\n\n    async fn emit_batches(\n        &mut self,\n        source_sink: &SourceSink,\n        ctx: &SourceContext,\n    ) -> Result<Duration, ActorExitStatus> {\n        let now = Instant::now();\n        let mut batch_builder = BatchBuilder::new(SourceType::Kafka);\n        let deadline = time::sleep(*EMIT_BATCHES_TIMEOUT);\n        tokio::pin!(deadline);\n\n        loop {\n            tokio::select! {\n                event_opt = self.events_rx.recv() => {\n                    let event = event_opt.ok_or_else(|| ActorExitStatus::from(anyhow!(\"consumer was dropped\")))?;\n                    match event {\n                        KafkaEvent::Message(message) => self.process_message(message, &mut batch_builder).await?,\n                        KafkaEvent::AssignPartitions { partitions, assignment_tx} => self.process_assign_partitions(ctx, &partitions, assignment_tx).await?,\n                        KafkaEvent::RevokePartitions { ack_tx } => self.process_revoke_partitions(ctx, source_sink, &mut batch_builder, ack_tx).await?,\n                        KafkaEvent::PartitionEOF(partition) => self.process_partition_eof(partition),\n                        KafkaEvent::Error(error) => Err(ActorExitStatus::from(error))?,\n                    }\n                    if batch_builder.num_bytes >= BATCH_NUM_BYTES_LIMIT {\n                        break;\n                    }\n                }\n                _ = &mut deadline => {\n                    break;\n                }\n            }\n            ctx.record_progress();\n        }\n        if !batch_builder.checkpoint_delta.is_empty() {","sourceCodeStart":450,"sourceCodeEnd":486,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/kafka_source.rs#L450-L486","documentation":"The Kafka source actor listens on an `events_rx` channel fed by a background consumer loop. `recv()` returning `None` means the sender (consumer) side was dropped, so no further Kafka events can arrive; the actor converts this into a fatal `ActorExitStatus` with this message.","triggerScenarios":"`emit_batches` selects on `self.events_rx.recv()` and the channel is closed because the Kafka consumer task/thread owning the sender exited or was dropped — e.g. consumer loop panicked, spawning failed, or the consumer was torn down while the actor kept running.","commonSituations":"Kafka consumer thread panic (deserialization error, broker auth crash), shutdown ordering where the consumer is dropped before the actor, or resource exhaustion killing the consumer task.","solutions":["Look upstream in the logs for the consumer task's panic or exit cause and fix that root error (auth, broker connectivity, deserialization).","Restart the source/indexer so the consumer and actor are re-spawned together.","Ensure broker connectivity and credentials (`bootstrap_servers`, SASL/TLS settings) are valid so the consumer loop survives.","If caused by shutdown ordering, this is benign during stop; otherwise report as a lifecycle bug."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match result {\n    Err(e) if e.to_string().contains(\"consumer was dropped\") => {\n        // find consumer task's root failure in logs, then restart the source actor\n    }\n    other => other?,\n}","preventionTips":["Validate Kafka credentials, TLS, and bootstrap server settings before deploying.","Watch for consumer-task panics (deserialization, OOM) in logs.","Ensure graceful shutdown drops the actor before/with the consumer to distinguish intentional stops.","Alert on repeated occurrences — each means a consumer died unexpectedly."],"tags":["kafka","source","actor","channel"],"backgroundTag":"connection-closed","analyzedSha":"a39730c5cdcd1a4fe798403737ae293999ea21f8","analyzedAt":"2026-09-08T13:19:37.784Z","contentChangedAt":"2026-09-08T13:19:37.784Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}