{"record":{"id":"301b52d0122cf692","repo":"quickwit-oss/quickwit","slug":"received-message-from-unassigned-partition-a","errorCode":null,"errorMessage":"received message from unassigned partition `{}`. Assigned partitions: `{{{}}}`","messagePattern":"received message from unassigned partition `(.+?)`\\. Assigned partitions: `(.+?)\\}\\}`","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/kafka_source.rs","lineNumber":315,"sourceCode":"            partition,\n            offset,\n            ..\n        } = message;\n\n        if let Some(doc) = doc_opt {\n            batch.add_doc(doc);\n        } else {\n            self.state.num_invalid_messages += 1;\n        }\n        self.state.num_bytes_processed += payload_len;\n        self.state.num_messages_processed += 1;\n\n        let partition_id = self\n            .state\n            .assigned_partitions\n            .get(&partition)\n            .ok_or_else(|| {\n                anyhow::anyhow!(\n                    \"received message from unassigned partition `{}`. Assigned partitions: \\\n                     `{{{}}}`\",\n                    partition,\n                    self.state.assigned_partitions.keys().join(\", \"),\n                )\n            })?\n            .clone();\n        let current_position = Position::offset(offset);\n        let previous_position = self\n            .state\n            .current_positions\n            .insert(partition, current_position.clone())\n            .unwrap_or_else(|| previous_position_for_offset(offset));\n        batch\n            .checkpoint_delta\n            .record_partition_delta(partition_id, previous_position, current_position)\n            .context(\"failed to record partition delta\")?;\n        Ok(())","sourceCodeStart":297,"sourceCodeEnd":333,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/kafka_source.rs#L297-L333","documentation":"The Kafka source actor only expects messages from partitions it was explicitly assigned via `AssignPartitions` events. If a message arrives for a partition absent from `assigned_partitions`, the actor treats it as a programming/consistency error and aborts with this message that also lists the currently assigned partitions.","triggerScenarios":"`process_message` receives a `KafkaEvent::Message` whose partition is not in `self.state.assigned_partitions` — e.g. events from a pre-revocation consumer racing with a `RevokePartitions`, duplicate assignment events, or a misbehaving consumer thread delivering stale records.","commonSituations":"Kafka consumer group rebalances racing with in-flight polled messages, restarting sources with cached consumer state, or broker/session timeouts triggering rapid partition revocation while buffered messages are still processed.","solutions":["Check consumer group rebalance frequency (session timeouts, `max.poll.interval.ms`) and stabilize membership to reduce revocation races.","Upgrade/verify Quickwit's kafka_source version for fixes to assignment handling during rebalances.","Ensure only one consumer instance per source assignment and no duplicate consumer configs.","Restart the indexing source node to re-establish a clean partition assignment."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match result {\n    Err(e) if e.to_string().contains(\"received message from unassigned partition\") => {\n        // restart the source to rebuild assignment; rebalances are transient\n    }\n    other => other?,\n}","preventionTips":["Tune consumer group settings (session timeout, max.poll.interval) to avoid churn-induced rebalances.","Avoid running duplicate consumers under the same group/source.","Keep broker connectivity stable to prevent spurious revocations.","Monitor consumer group rebalance events alongside indexing logs."],"tags":["kafka","source","partitioning","rebalance"],"backgroundTag":"invalid-state-transition","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"}