{"record":{"id":"17285f9f0bd9c3b3","repo":"quickwit-oss/quickwit","slug":"consumer-was-dropped-17285f","errorCode":null,"errorMessage":"consumer was dropped","messagePattern":"consumer was dropped","errorType":"exception","errorClass":"ActorExitStatus","httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/pulsar_source.rs","lineNumber":230,"sourceCode":"impl Source for PulsarSource {\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::Pulsar);\n        let deadline = time::sleep(*EMIT_BATCHES_TIMEOUT);\n        tokio::pin!(deadline);\n\n        loop {\n            tokio::select! {\n                // This does not actually acquire the lock of the mutex internally\n                // we're using the mutex in order to convince the Rust compiler\n                // that we can use the consumer within this Sync context.\n                message = self.pulsar_consumer.next() => {\n                    let message = message\n                        .ok_or_else(|| ActorExitStatus::from(anyhow!(\"consumer was dropped\")))?\n                        .map_err(|e| ActorExitStatus::from(anyhow!(\"failed to get message from consumer: {:?}\", e)))?;\n\n                    self.process_message(message, &mut batch_builder).map_err(ActorExitStatus::from)?;\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\n        if !batch_builder.checkpoint_delta.is_empty() {\n            debug!(\n                num_docs=%batch_builder.docs.len(),","sourceCodeStart":212,"sourceCodeEnd":248,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/pulsar_source.rs#L212-L248","documentation":"The Pulsar source actor's `emit_batches` loop polls `pulsar_consumer.next()`, which returns None only when the consumer object has been dropped/closed. Since the source requires a live consumer to function, it converts that None into an ActorExitStatus and terminates. This is the graceful signal that the consumer can no longer deliver messages.","triggerScenarios":"`self.pulsar_consumer.next()` (the Stream::next future) resolves to None, which happens when the consumer instance has been dropped or closed — e.g. the consumer was shut down while the actor still held it in the mutex, or the consumer channel was closed by the Pulsar client.","commonSituations":"Pulsar client/connection teardown racing with the actor loop; the consumer being closed elsewhere in the code; a failed subscription that caused the client to drop the consumer internally.","solutions":["Check why the consumer was dropped before the actor exited — look for code paths that close or drop the Pulsar consumer early.","Verify the Pulsar client and connection remain alive for the lifetime of the source actor (broker restarts, connection loss).","Restart the indexing source so a fresh consumer is created against the topic.","If caused by broker-side unavailability, ensure the Pulsar broker/service URL is reachable and the topic/subscription exists."],"exampleFix":"// before\nlet message = message.ok_or_else(|| ActorExitStatus::from(anyhow!(\"consumer was dropped\")))?;\n// after: recreate consumer instead of exiting\nlet message = match message {\n    Some(msg) => msg,\n    None => {\n        warn!(\"pulsar consumer dropped; recreating\");\n        self.recreate_consumer(ctx).await?;\n        continue;\n    }\n};","handlingStrategy":"retry","validationCode":"// before starting the actor, confirm the consumer is live\nassert!(!consumer_is_closed(&pulsar_consumer), \"pulsar consumer closed before source start\");","typeGuard":null,"tryCatchPattern":"match source_future.await {\n    Err(e) if e.to_string().contains(\"consumer was dropped\") => {\n        warn!(\"pulsar consumer dropped; recreating source\");\n        spawn_pulsar_source(cfg).await?;\n    }\n    other => other?,\n}","preventionTips":["Keep the Pulsar client alive for the entire actor lifetime (store it alongside the consumer).","Monitor broker connectivity and restart the source on consumer loss.","Avoid closing the consumer from other tasks while the source is running."],"tags":["pulsar","consumer","actor","streaming"],"backgroundTag":"internal-invariant-violation","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"}