quickwit-oss/quickwit · error · ActorExitStatus

consumer was dropped

Error message

consumer was dropped

What it means

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.

Source

Thrown at quickwit/quickwit-indexing/src/source/pulsar_source.rs:230

impl Source for PulsarSource {
    async fn emit_batches(
        &mut self,
        source_sink: &SourceSink,
        ctx: &SourceContext,
    ) -> Result<Duration, ActorExitStatus> {
        let now = Instant::now();
        let mut batch_builder = BatchBuilder::new(SourceType::Pulsar);
        let deadline = time::sleep(*EMIT_BATCHES_TIMEOUT);
        tokio::pin!(deadline);

        loop {
            tokio::select! {
                // This does not actually acquire the lock of the mutex internally
                // we're using the mutex in order to convince the Rust compiler
                // that we can use the consumer within this Sync context.
                message = self.pulsar_consumer.next() => {
                    let message = message
                        .ok_or_else(|| ActorExitStatus::from(anyhow!("consumer was dropped")))?
                        .map_err(|e| ActorExitStatus::from(anyhow!("failed to get message from consumer: {:?}", e)))?;

                    self.process_message(message, &mut batch_builder).map_err(ActorExitStatus::from)?;

                    if batch_builder.num_bytes >= BATCH_NUM_BYTES_LIMIT {
                        break;
                    }
                }
                _ = &mut deadline => {
                    break;
                }
            }
            ctx.record_progress();
        }

        if !batch_builder.checkpoint_delta.is_empty() {
            debug!(
                num_docs=%batch_builder.docs.len(),

View on GitHub (pinned to a39730c5cd)

Solutions

1. Check why the consumer was dropped before the actor exited — look for code paths that close or drop the Pulsar consumer early.
2. Verify the Pulsar client and connection remain alive for the lifetime of the source actor (broker restarts, connection loss).
3. Restart the indexing source so a fresh consumer is created against the topic.
4. If caused by broker-side unavailability, ensure the Pulsar broker/service URL is reachable and the topic/subscription exists.

Example fix

// before
let message = message.ok_or_else(|| ActorExitStatus::from(anyhow!("consumer was dropped")))?;
// after: recreate consumer instead of exiting
let message = match message {
    Some(msg) => msg,
    None => {
        warn!("pulsar consumer dropped; recreating");
        self.recreate_consumer(ctx).await?;
        continue;
    }
};

// before starting the actor, confirm the consumer is live
assert!(!consumer_is_closed(&pulsar_consumer), "pulsar consumer closed before source start");

match source_future.await {
    Err(e) if e.to_string().contains("consumer was dropped") => {
        warn!("pulsar consumer dropped; recreating source");
        spawn_pulsar_source(cfg).await?;
    }
    other => other?,
}

When it happens

Trigger: `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.

Common situations: 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.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• failed to get message from consumer: {:?}
• actor `{}` is disconnected
• Quickwit was compiled without the `pulsar` feature
• merge requires at least one input
• consumer was dropped

AI-assisted analysis of quickwit-oss/quickwit@a39730c5cd (2026-09-08). Data as JSON: /api/errors/17285f9f0bd9c3b3. Report an issue: GitHub.