risingwavelabs/risingwave · error · SinkError::Kinesis
failed to send records. sent {} out of {}, last err: code: [
Error message
failed to send records. sent {} out of {}, last err: code: [{}], message: [{}] What it means
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.
Source
Thrown at src/connector/src/sink/kinesis.rs:333
);
start_idx += partially_sent_count;
// reset retry count when having progress
remaining_no_progress_retry_count = MAX_NO_PROGRESS_RETRY_COUNT;
} else if let Some(err_code) = &result_entry.error_code && err_code == "ProvisionedThroughputExceededException" {
// From the doc of `put_records`:
// The ErrorCode parameter reflects the type of error and can be one of the following values:
// ProvisionedThroughputExceededException or InternalFailure. ErrorMessage provides more detailed
// information about the ProvisionedThroughputExceededException exception including the account ID,
// stream name, and shard ID of the record that was throttled.
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");
warn!(err_string = ?result_entry.error_message, ?throttle_delay, "throttle");
sleep(throttle_delay).await;
} else {
// no progress due to some internal error
assert_eq!(first_failed_idx, 0);
remaining_no_progress_retry_count -= 1;
if remaining_no_progress_retry_count == 0 {
return Err(SinkError::Kinesis(anyhow!(
"failed to send records. sent {} out of {}, last err: code: [{}], message: [{}]",
start_idx,
total_count,
result_entry.error_code.unwrap_or_default(),
result_entry.error_message.unwrap_or_default()
)));
} else {
warn!(
remaining_no_progress_retry_count,
sent = start_idx,
total_count,
"failed to send records. code: [{}], message: [{}]",
result_entry.error_code.unwrap_or_default(),
result_entry.error_message.unwrap_or_default()
)
}
}
} else {View on GitHub (pinned to 6469eb736d)
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
Defensive patterns
Strategy: retry
Validate before calling
aws kinesis describe-stream --stream-name s1 # confirm StreamStatus == ACTIVE before starting the sink
Try / catch
match sink.finish().await {
Err(e) if e.to_string().contains("failed to send records") => {
// parse 'sent X out of Y' for progress; resume from checkpoint on retry
error!("kinesis sink gave up: {e:#}");
schedule_checkpoint_restart();
Err(e)
}
r => r,
} Prevention
- 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)
When it happens
Trigger: 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`.
Common situations: Stream deleted or in a bad state; InternalFailure on the AWS side; invalid stream name persistently rejected; sustained endpoint malfunction with custom endpoints.
Related errors
- request record count {} not match the response record count
- failed to send records. sent {} out of {}
- failed to write {} unprocessed items to DynamoDB sink after
- failed to write items to DynamoDB sink
- missing FORMAT ... ENCODE ...
AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11).
Data as JSON: /api/errors/be4c10a606236d32.
Report an issue: GitHub.