{"record":{"id":"be4c10a606236d32","repo":"risingwavelabs/risingwave","slug":"failed-to-send-records-sent-out-of-last-er","errorCode":null,"errorMessage":"failed to send records. sent {} out of {}, last err: code: [{}], message: [{}]","messagePattern":"failed to send records\\. sent (.+?) out of (.+?), last err: code: \\[(.+?)\\], message: \\[(.+?)\\]","errorType":"exception","errorClass":"SinkError::Kinesis","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":333,"sourceCode":"                                    );\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,\n                                            total_count,\n                                            \"failed to send records. code: [{}], message: [{}]\",\n                                            result_entry.error_code.unwrap_or_default(),\n                                            result_entry.error_message.unwrap_or_default()\n                                        )\n                                    }\n                                }\n                            } else {","sourceCodeStart":315,"sourceCodeEnd":351,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L315-L351","documentation":"After exhausting no-progress retries, the sink gives up with a summary error reporting how many records were sent, plus the error code/message from the last failed record entry. It fires when PutRecords keeps returning per-record failures (other than throughput throttling) and the remaining no-progress retry budget reaches zero.","triggerScenarios":"Repeated PutRecords responses where the first entry has a persistent error_code (e.g. InternalFailure) so `remaining_no_progress_retry_count` hits 0 in `finish`.","commonSituations":"Stream deleted or in a bad state; InternalFailure on the AWS side; invalid stream name persistently rejected; sustained endpoint malfunction with custom endpoints.","solutions":["Verify the stream name and that the stream exists and is ACTIVE","Inspect result_entry.error_code in the message to identify the underlying AWS error","Retry the sink; resume from checkpoint so already-sent records are not duplicated","Check AWS service health / endpoint reachability if using a custom endpoint"],"exampleFix":null,"handlingStrategy":"retry","validationCode":"aws kinesis describe-stream --stream-name s1\n# confirm StreamStatus == ACTIVE before starting the sink","typeGuard":null,"tryCatchPattern":"match sink.finish().await {\n  Err(e) if e.to_string().contains(\"failed to send records\") => {\n    // parse 'sent X out of Y' for progress; resume from checkpoint on retry\n    error!(\"kinesis sink gave up: {e:#}\");\n    schedule_checkpoint_restart();\n    Err(e)\n  }\n  r => r,\n}","preventionTips":["Verify stream exists and is ACTIVE before creating the sink","Check the embedded error_code for the real AWS failure","Ensure stream name matches exactly (region-scoped)"],"tags":["kinesis","put-records","retry-exhausted","write-failure"],"backgroundTag":"api-error-response","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"}