{"record":{"id":"a156cb68fda3132a","repo":"quickwit-oss/quickwit","slug":"failed-to-get-message-from-consumer","errorCode":null,"errorMessage":"failed to get message from consumer: {:?}","messagePattern":"failed to get message from consumer: (.+?)","errorType":"exception","errorClass":"ActorExitStatus","httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/pulsar_source.rs","lineNumber":231,"sourceCode":"    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(),\n                num_bytes=%batch_builder.num_bytes,","sourceCodeStart":213,"sourceCodeEnd":249,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/pulsar_source.rs#L213-L249","documentation":"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.","triggerScenarios":"`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`.","commonSituations":"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.","solutions":["Inspect the wrapped error (`{:?}` payload in the message) for the underlying Pulsar error cause.","Verify connectivity to the Pulsar broker (service URL, network, DNS) and that the topic/subscription still exist and permissions are valid.","Check authentication credentials (token/JWT) haven't expired; refresh them.","Restart the source actor; the Pulsar client typically reconnects on its own once the broker is reachable."],"exampleFix":"// before\n.map_err(|e| ActorExitStatus::from(anyhow!(\"failed to get message from consumer: {:?}\", e)))?;\n// after: log the underlying error for diagnostics\n.map_err(|e| {\n    error!(error = ?e, \"pulsar receive failed\");\n    ActorExitStatus::from(anyhow!(\"failed to get message from consumer: {:?}\", e))\n})?;","handlingStrategy":"retry","validationCode":"// pre-flight: verify broker reachability and topic access\npulsar_client.lookup_topic(&topic_url).await\n    .context(\"pulsar broker unreachable or topic missing before source start\")?;","typeGuard":null,"tryCatchPattern":"match source_future.await {\n    Err(e) if e.to_string().contains(\"failed to get message from consumer\") => {\n        warn!(error = %e, \"pulsar receive failed; restarting source with backoff\");\n        backoff(|| spawn_pulsar_source(cfg)).await?;\n    }\n    other => other?,\n}","preventionTips":["Verify Pulsar service URL, credentials, and topic permissions before deployment.","Monitor auth token expiry and rotate tokens before they lapse.","Add retries with backoff around source restarts to survive broker restarts and network blips."],"tags":["pulsar","consumer","network","streaming"],"backgroundTag":"upstream-api-error","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"}