risingwavelabs/risingwave · error · ConnectorError

Kinesis got an unhandled error on stream {:?}, shard {:?}

Error message

Kinesis got an unhandled error on stream {:?}, shard {:?}

What it means

In `into_data_stream` (invoked from `into_stream`), when reading records from a Kinesis shard returns an error that is not one of the specially handled (retryable) cases, the code wraps it with stream/shard context via `anyhow!` and `context(format!(...))` and returns it, terminating the shard reader. It signals an unrecognized Kinesis client failure for this stream/shard.

Source

Thrown at src/connector/src/source/kinesis/source/reader.rs:316

                    self.new_shard_iter().await?;
                    tokio::time::sleep(self.error_retry_interval).await;
                    continue;
                }
                Err(SdkError::TimeoutError(_)) => {
                    self.kinesis_timeout_count.inc();

                    // according to sdk doc:
                    // The request failed due to a timeout. The request MAY have been sent and received.
                    tracing::warn!(
                        "shard {:?} request timeout, rolling back to previous offset",
                        self.shard_id
                    );
                    self.new_shard_iter().await?;
                    tokio::time::sleep(self.error_retry_interval).await;
                    continue;
                }
                Err(e) => {
                    let error = anyhow!(e).context(format!(
                        "Kinesis got an unhandled error on stream {:?}, shard {:?}",
                        self.stream_name, self.shard_id
                    ));
                    tracing::warn!(error = %error.as_report()); // change to warn as user has no action to take
                    return Err(error.into());
                }
            }
        }
    }
}
impl KinesisSplitReader {
    async fn new_shard_iter(&mut self) -> Result<()> {
        let (starting_seq_num, start_timestamp, iter_type) = if self.latest_offset.is_some() {
            (
                self.latest_offset.clone(),
                None,
                ShardIteratorType::AfterSequenceNumber,
            )

View on GitHub (pinned to 6469eb736d)

Solutions

1. Check the wrapped cause in the error report (AWS SDK error code) to identify the underlying failure.
2. For ProvisionedThroughputExceeded-type issues, increase shards or reduce consumers reading the shard.
3. Verify IAM credentials/policy for kinesis:ListShard/GetShardIterator/GetRecords are still valid.
4. Restart the source; the connector retries from the persisted offset.
5. If the error is persistently misclassified as unhandled (e.g. expired iterator), report/patch the retry match in reader.rs to handle it.

// The source returns anyhow::Error with context; inspect the chain for the AWS SDK error kind
match res {
    Err(e) if e.to_string().contains("ProvisionedThroughputExceeded") || e.to_string().contains("ExpiredIteratorException") => {
        // backoff and restart the source
        tokio::time::sleep(Duration::from_secs(5)).await;
        restart_source(source_id).await?;
    }
    Err(e) => { tracing::error!("kinesis source failed: {}", e.as_report()); alert(); }
    Ok(()) => {}
}

When it happens

Trigger: Any unhandled AWS SDK error during `get_records` on a shard: expired shard iterator treated as non-retryable by the match, throttling/limit-exceeded variants not covered by earlier arms, credential or permission failures surfacing mid-stream, SDK dispatch/network errors outside the handled set.

Common situations: AWS Kinesis throttling under high shard throughput, IAM policy changes revoking kinesis:GetRecords mid-run, long-lived iterators expiring on idle shards, transient network partitions to AWS endpoints.

Related errors

• error building Kinesis client: {err}
• should not fail because we have set `data` and `partition_ke
• shard iterator is none
• missing FORMAT ... ENCODE ...
• kinesis sink requires partition key (please define in `prima

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/4eec752045d04726. Report an issue: GitHub.