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.

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
• attempted to merge zero splits
• body stream ended after skipping
• body stream ended with
• consumer was dropped

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

Appendix: 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)