quickwit-oss/quickwit · error · ActorExitStatus

failed to get message from consumer: {:?}

Error message

failed to get message from consumer: {:?}

What it means

Wraps an error returned by the Pulsar consumer while fetching the next message. The source actor converts the underlying consumer error into an ActorExitStatus after formatting it with debug formatting, so the root cause appears in the exit status. It means the receive operation itself failed, not that the stream was closed.

Source

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

    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(),
                num_bytes=%batch_builder.num_bytes,

View on GitHub (pinned to a39730c5cd)

Solutions

  1. Inspect the wrapped error (`{:?}` payload in the message) for the underlying Pulsar error cause.
  2. Verify connectivity to the Pulsar broker (service URL, network, DNS) and that the topic/subscription still exist and permissions are valid.
  3. Check authentication credentials (token/JWT) haven't expired; refresh them.
  4. Restart the source actor; the Pulsar client typically reconnects on its own once the broker is reachable.

Example fix

// before
.map_err(|e| ActorExitStatus::from(anyhow!("failed to get message from consumer: {:?}", e)))?;
// after: log the underlying error for diagnostics
.map_err(|e| {
    error!(error = ?e, "pulsar receive failed");
    ActorExitStatus::from(anyhow!("failed to get message from consumer: {:?}", e))
})?;
Defensive patterns

Strategy: retry

Validate before calling

// pre-flight: verify broker reachability and topic access
pulsar_client.lookup_topic(&topic_url).await
    .context("pulsar broker unreachable or topic missing before source start")?;

Try / catch

match source_future.await {
    Err(e) if e.to_string().contains("failed to get message from consumer") => {
        warn!(error = %e, "pulsar receive failed; restarting source with backoff");
        backoff(|| spawn_pulsar_source(cfg)).await?;
    }
    other => other?,
}

Prevention

When it happens

Trigger: `pulsar_consumer.next()` yields `Some(Err(e))` — e.g. connection failure to the Pulsar broker, subscription lookup failure, deserialization error, or broker-side error while the actor is building its batch in `emit_batches`.

Common situations: Pulsar broker restart or network partition mid-consumption; authentication token expired; topic deleted or permissions revoked while running; consumer in a failed state after a protocol error.

Related errors

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