risingwavelabs/risingwave · warning

should not be none

Error message

should not be none

What it means

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.

Source

Thrown at src/connector/src/sink/kinesis.rs:325

                                let partially_sent_count = first_failed_idx;
                                if partially_sent_count > 0 {
                                    warn!(
                                        partially_sent_count,
                                        record_count,
                                        "records are partially sent. code: [{}], message: [{}]",
                                        result_entry.error_code.unwrap_or_default(),
                                        result_entry.error_message.unwrap_or_default()
                                    );
                                    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,

View on GitHub (pinned to 6469eb736d)

Solutions

1. This is an internal assert, not user-facing; if it panics, report a bug in the backoff helper
2. Raise the stream's provisioned throughput or switch the stream to on-demand capacity mode
3. Reduce sink parallelism/write rate or spread writes across more shards via better partition keys

// pre-check stream capacity expectations
// ensure the Kinesis stream is ACTIVE and sized for your write rate
aws kinesis describe-stream-summary --stream-name s1
# check StreamStatus == ACTIVE and ShardCount / on-demand mode

// throttling is auto-retried with backoff internally; only the internal expect is fatal
match sink.finish().await {
  Err(e) => { warn!("kinesis write failed: {e:#}"); Err(e) }
  ok => ok,
}

When it happens

Trigger: Hitting shard throttling during `finish` retry loop, triggering the `get_or_insert_with(...).next().expect("should not be none")` path.

Common situations: Writing to an under-provisioned Kinesis stream (on-demand disabled, shard count too low); burst traffic exceeding shard limits.

Related errors

• should not fail because we have set `data` and `partition_ke
• failed to write {} unprocessed items to DynamoDB sink after
• missing FORMAT ... ENCODE ...
• kinesis sink requires partition key (please define in `prima
• serde (de)serialization error for KinesisSinkConfig: {e}

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