{"record":{"id":"9f9f0c84d3fba99d","repo":"risingwavelabs/risingwave","slug":"should-not-be-none","errorCode":null,"errorMessage":"should not be none","messagePattern":"should not be none","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"warning","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":325,"sourceCode":"                                let partially_sent_count = first_failed_idx;\n                                if partially_sent_count > 0 {\n                                    warn!(\n                                        partially_sent_count,\n                                        record_count,\n                                        \"records are partially sent. code: [{}], message: [{}]\",\n                                        result_entry.error_code.unwrap_or_default(),\n                                        result_entry.error_message.unwrap_or_default()\n                                    );\n                                    start_idx += partially_sent_count;\n                                    // reset retry count when having progress\n                                    remaining_no_progress_retry_count = MAX_NO_PROGRESS_RETRY_COUNT;\n                                } else if let Some(err_code) = &result_entry.error_code && err_code == \"ProvisionedThroughputExceededException\" {\n                                    // From the doc of `put_records`:\n                                    // The ErrorCode parameter reflects the type of error and can be one of the following values:\n                                    // ProvisionedThroughputExceededException or InternalFailure. ErrorMessage provides more detailed\n                                    // information about the ProvisionedThroughputExceededException exception including the account ID,\n                                    // stream name, and shard ID of the record that was throttled.\n                                    let throttle_delay = throttle_delay.get_or_insert_with(|| exponential_backoff(Duration::from_millis(100), 2, Duration::from_secs(2)).map(jitter)).next().expect(\"should not be none\");\n                                    warn!(err_string = ?result_entry.error_message, ?throttle_delay, \"throttle\");\n                                    sleep(throttle_delay).await;\n                                } else  {\n                                    // no progress due to some internal error\n                                    assert_eq!(first_failed_idx, 0);\n                                    remaining_no_progress_retry_count -= 1;\n                                    if remaining_no_progress_retry_count == 0 {\n                                        return Err(SinkError::Kinesis(anyhow!(\n                                            \"failed to send records. sent {} out of {}, last err: code: [{}], message: [{}]\",\n                                            start_idx,\n                                            total_count,\n                                            result_entry.error_code.unwrap_or_default(),\n                                            result_entry.error_message.unwrap_or_default()\n                                        )));\n                                    } else {\n                                        warn!(\n                                            remaining_no_progress_retry_count,\n                                            sent = start_idx,","sourceCodeStart":307,"sourceCodeEnd":343,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L307-L343","documentation":"When records fail with ProvisionedThroughputExceededException, the sink lazily creates an exponential backoff iterator and unwraps its first delay with expect(). The unwrap is an internal invariant: the backoff generator is configured with factor 2 and a max delay, so its first item always exists. It only fires if the generator logic is broken.","triggerScenarios":"Hitting shard throttling during `finish` retry loop, triggering the `get_or_insert_with(...).next().expect(\"should not be none\")` path.","commonSituations":"Writing to an under-provisioned Kinesis stream (on-demand disabled, shard count too low); burst traffic exceeding shard limits.","solutions":["This is an internal assert, not user-facing; if it panics, report a bug in the backoff helper","Raise the stream's provisioned throughput or switch the stream to on-demand capacity mode","Reduce sink parallelism/write rate or spread writes across more shards via better partition keys"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"// pre-check stream capacity expectations\n// ensure the Kinesis stream is ACTIVE and sized for your write rate\naws kinesis describe-stream-summary --stream-name s1\n# check StreamStatus == ACTIVE and ShardCount / on-demand mode","typeGuard":null,"tryCatchPattern":"// throttling is auto-retried with backoff internally; only the internal expect is fatal\nmatch sink.finish().await {\n  Err(e) => { warn!(\"kinesis write failed: {e:#}\"); Err(e) }\n  ok => ok,\n}","preventionTips":["Provision shards or enable on-demand mode to match sink throughput","Monitor ProvisionedThroughputExceededException metrics in CloudWatch","Keep partition keys well-distributed to avoid hot shards"],"tags":["kinesis","throttling","backoff","assert"],"backgroundTag":"rate-limit-exceeded","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"}