{"record":{"id":"dd9c2747122c91ef","repo":"quickwit-oss/quickwit","slug":"received-record-from-unassigned-shard","errorCode":null,"errorMessage":"received record from unassigned shard `{}`","messagePattern":"received record from unassigned shard `(.+?)`","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/kinesis/kinesis_source.rs","lineNumber":263,"sourceCode":"                                if record_data.is_empty() {\n                                    warn!(\n                                        stream_name=%self.stream_name,\n                                        shard_id=%shard_id,\n                                        sequence_number=%record.sequence_number,\n                                        \"record is empty\"\n                                    );\n                                    self.state.num_invalid_records += 1;\n                                    continue;\n                                }\n                                batch_builder.add_doc(Bytes::from(record_data));\n\n                                if i == num_records - 1 {\n                                    let shard_consumer_state = self\n                                        .state\n                                        .shard_consumers\n                                        .get_mut(&shard_id)\n                                        .ok_or_else(|| {\n                                            anyhow::anyhow!(\n                                                \"received record from unassigned shard `{}`\", shard_id,\n                                            )\n                                        })?;\n                                    shard_consumer_state.lag_millis = lag_millis;\n\n                                    let partition_id = shard_consumer_state.partition_id.clone();\n                                    let current_position = Position::from(record.sequence_number);\n                                    let previous_position = std::mem::replace(&mut shard_consumer_state.current_position, current_position.clone());\n\n                                    batch_builder.checkpoint_delta.record_partition_delta(\n                                        partition_id,\n                                        previous_position,\n                                        current_position,\n                                    ).context(\"failed to record partition delta\")?;\n                                }\n                            }\n                            if batch_builder.num_bytes >= BATCH_NUM_BYTES_LIMIT {\n                                break;","sourceCodeStart":245,"sourceCodeEnd":281,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/kinesis/kinesis_source.rs#L245-L281","documentation":"This error is thrown when a record arrives from a Kinesis shard that the source actor has no consumer state for. Each record batch's last record is used to update the shard consumer's lag; if the shard_id is not present in `state.shard_consumers`, the internal bookkeeping and the external world disagree, so the actor fails fast instead of silently dropping the record. It is an internal invariant violation indicating stale or corrupted shard-assignment state.","triggerScenarios":"A Kinesis record is received whose shard_id has no entry in the actor's `shard_consumers` map — typically after shard assignment changed (shard split/merge, lease reassignment, or a restart with a stale Kinesis iterator) but the source kept consuming from the old stream position.","commonSituations":"Resharding of the Kinesis stream while the source is running; an indexing pipeline restarted with stale checkpoint data pointing at shards that were merged or trimmed; bugs or race conditions in shard assignment so the consumer pulls a shard it was never assigned.","solutions":["Restart the indexing source so shard assignment state is rebuilt from the current stream topology.","Verify the Kinesis stream has not been resharded (split/merged shards) while the source was running; recreate the source for the new stream layout.","Check that the checkpointed shard iterator / position still corresponds to shards in the assigned set and fix stale checkpoint data.","If reproducible, file a bug: the shard consumer initialization and record consumption paths are out of sync."],"exampleFix":"// before (stale checkpoint consumed from removed shard)\nlet shard_consumer_state = self.state.shard_consumers.get_mut(&shard_id).ok_or_else(...)?;\n// after: verify assignment before consuming\nif !self.state.shard_consumers.contains_key(&shard_id) {\n    warn!(shard_id = %shard_id, \"skipping record from unassigned shard\");\n    continue;\n}","handlingStrategy":"validation","validationCode":"if !source_state.shard_consumers.contains_key(&shard_id) {\n    return Err(anyhow!(\"shard {} is not assigned to this source; restart the source\", shard_id));\n}","typeGuard":"fn is_shard_assigned(state: &SourceState, shard_id: &ShardId) -> bool {\n    state.shard_consumers.contains_key(shard_id)\n}","tryCatchPattern":"match result {\n    Err(e) if e.to_string().contains(\"received record from unassigned shard\") => {\n        warn!(error = %e, \"stale shard assignment; restarting source\");\n        restart_source().await?;\n    }\n    other => other?,\n}","preventionTips":["Restart/rebuild the source after any Kinesis stream resharding operation.","Ensure shard assignment and record consumption derive from the same coordinated state snapshot.","Alert on this error in production: it signals stale checkpoint or assignment state, not a transient issue."],"tags":["kinesis","sharding","invariant","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-14T11:17:12.474Z"}