{"record":{"id":"4eec752045d04726","repo":"risingwavelabs/risingwave","slug":"kinesis-got-an-unhandled-error-on-stream-sha","errorCode":null,"errorMessage":"Kinesis got an unhandled error on stream {:?}, shard {:?}","messagePattern":"Kinesis got an unhandled error on stream (.+?), shard (.+?)","errorType":"exception","errorClass":"ConnectorError","httpStatus":null,"severity":"error","filePath":"src/connector/src/source/kinesis/source/reader.rs","lineNumber":316,"sourceCode":"                    self.new_shard_iter().await?;\n                    tokio::time::sleep(self.error_retry_interval).await;\n                    continue;\n                }\n                Err(SdkError::TimeoutError(_)) => {\n                    self.kinesis_timeout_count.inc();\n\n                    // according to sdk doc:\n                    // The request failed due to a timeout. The request MAY have been sent and received.\n                    tracing::warn!(\n                        \"shard {:?} request timeout, rolling back to previous offset\",\n                        self.shard_id\n                    );\n                    self.new_shard_iter().await?;\n                    tokio::time::sleep(self.error_retry_interval).await;\n                    continue;\n                }\n                Err(e) => {\n                    let error = anyhow!(e).context(format!(\n                        \"Kinesis got an unhandled error on stream {:?}, shard {:?}\",\n                        self.stream_name, self.shard_id\n                    ));\n                    tracing::warn!(error = %error.as_report()); // change to warn as user has no action to take\n                    return Err(error.into());\n                }\n            }\n        }\n    }\n}\nimpl KinesisSplitReader {\n    async fn new_shard_iter(&mut self) -> Result<()> {\n        let (starting_seq_num, start_timestamp, iter_type) = if self.latest_offset.is_some() {\n            (\n                self.latest_offset.clone(),\n                None,\n                ShardIteratorType::AfterSequenceNumber,\n            )","sourceCodeStart":298,"sourceCodeEnd":334,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/source/kinesis/source/reader.rs#L298-L334","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Check the wrapped cause in the error report (AWS SDK error code) to identify the underlying failure.","For ProvisionedThroughputExceeded-type issues, increase shards or reduce consumers reading the shard.","Verify IAM credentials/policy for kinesis:ListShard/GetShardIterator/GetRecords are still valid.","Restart the source; the connector retries from the persisted offset.","If the error is persistently misclassified as unhandled (e.g. expired iterator), report/patch the retry match in reader.rs to handle it."],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// The source returns anyhow::Error with context; inspect the chain for the AWS SDK error kind\nmatch res {\n    Err(e) if e.to_string().contains(\"ProvisionedThroughputExceeded\") || e.to_string().contains(\"ExpiredIteratorException\") => {\n        // backoff and restart the source\n        tokio::time::sleep(Duration::from_secs(5)).await;\n        restart_source(source_id).await?;\n    }\n    Err(e) => { tracing::error!(\"kinesis source failed: {}\", e.as_report()); alert(); }\n    Ok(()) => {}\n}","preventionTips":["Provision enough shards / use on-demand mode to avoid throughput throttling.","Grant kinesis:ListShard, kinesis:GetShardIterator, kinesis:GetRecords to the runtime credentials and monitor IAM changes.","Keep active consumption so iterators do not expire from long idleness.","Monitor the error report's cause chain to distinguish retryable from permanent AWS errors."],"tags":["kinesis","aws-sdk","stream-read","unhandled-error"],"backgroundTag":"upstream-api-error","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}