quickwit-oss/quickwit · error · anyhow::Error

received record from unassigned shard `{}`

Error message

received record from unassigned shard `{}`

What it means

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.

Source

Thrown at quickwit/quickwit-indexing/src/source/kinesis/kinesis_source.rs:263

                                if record_data.is_empty() {
                                    warn!(
                                        stream_name=%self.stream_name,
                                        shard_id=%shard_id,
                                        sequence_number=%record.sequence_number,
                                        "record is empty"
                                    );
                                    self.state.num_invalid_records += 1;
                                    continue;
                                }
                                batch_builder.add_doc(Bytes::from(record_data));

                                if i == num_records - 1 {
                                    let shard_consumer_state = self
                                        .state
                                        .shard_consumers
                                        .get_mut(&shard_id)
                                        .ok_or_else(|| {
                                            anyhow::anyhow!(
                                                "received record from unassigned shard `{}`", shard_id,
                                            )
                                        })?;
                                    shard_consumer_state.lag_millis = lag_millis;

                                    let partition_id = shard_consumer_state.partition_id.clone();
                                    let current_position = Position::from(record.sequence_number);
                                    let previous_position = std::mem::replace(&mut shard_consumer_state.current_position, current_position.clone());

                                    batch_builder.checkpoint_delta.record_partition_delta(
                                        partition_id,
                                        previous_position,
                                        current_position,
                                    ).context("failed to record partition delta")?;
                                }
                            }
                            if batch_builder.num_bytes >= BATCH_NUM_BYTES_LIMIT {
                                break;

View on GitHub (pinned to a39730c5cd)

Solutions

1. Restart the indexing source so shard assignment state is rebuilt from the current stream topology.
2. Verify the Kinesis stream has not been resharded (split/merged shards) while the source was running; recreate the source for the new stream layout.
3. Check that the checkpointed shard iterator / position still corresponds to shards in the assigned set and fix stale checkpoint data.
4. If reproducible, file a bug: the shard consumer initialization and record consumption paths are out of sync.

Example fix

// before (stale checkpoint consumed from removed shard)
let shard_consumer_state = self.state.shard_consumers.get_mut(&shard_id).ok_or_else(...)?;
// after: verify assignment before consuming
if !self.state.shard_consumers.contains_key(&shard_id) {
    warn!(shard_id = %shard_id, "skipping record from unassigned shard");
    continue;
}

if !source_state.shard_consumers.contains_key(&shard_id) {
    return Err(anyhow!("shard {} is not assigned to this source; restart the source", shard_id));
}

fn is_shard_assigned(state: &SourceState, shard_id: &ShardId) -> bool {
    state.shard_consumers.contains_key(shard_id)
}

match result {
    Err(e) if e.to_string().contains("received record from unassigned shard") => {
        warn!(error = %e, "stale shard assignment; restarting source");
        restart_source().await?;
    }
    other => other?,
}

When it happens

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

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

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

• attempted to merge splits with different doc mapping uid
• unable to sniff region from environment
• Quickwit was compiled without the `kinesis` feature
• Partition is owned by this indexing pipeline but is not at t
• recovery time range must contain both start and end

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