risingwavelabs/risingwave · error · SinkError::Kinesis

request record count {} not match the response record count

Error message

request record count {} not match the response record count {}

What it means

After PutRecords succeeds, the sink expects the response to echo one entry per submitted record; Kinesis guarantees this per API contract. A mismatch indicates a violated invariant and is treated as an internal sink error rather than silently losing per-record results.

Source

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

                    }
                    if records.is_empty() {
                        // at least include one record even if its size exceed `MAX_TOTAL_RECORD_PAYLOAD_SIZE`
                        records.push(self.entries[start_idx].0.clone());
                    }

                    // 2. send the records and handle the result
                    let record_count = records.len();
                    match self
                        .client
                        .put_records()
                        .stream_name(&self.stream_name)
                        .set_records(Some(records))
                        .send()
                        .await
                    {
                        Ok(output) => {
                            if record_count != output.records.len() {
                                return Err(SinkError::Kinesis(anyhow!("request record count {} not match the response record count {}", record_count, output.records.len())));
                            }
                            // From the doc of `put_records`:
                            // A single record failure does not stop the processing of subsequent records. As a result,
                            // PutRecords doesn’t guarantee the ordering of records. If you need to read records in the same
                            // order they are written to the stream, use PutRecord instead of PutRecords, and write to the same shard.

                            // Therefore, to ensure at least once and eventual consistency, we figure out the first failed entry, and retry
                            // all the following entries even if the following entries may have been successfully processed.
                            if let Some((first_failed_idx, result_entry)) = Self::first_failed_entry(output) {
                                // first_failed_idx is also the number of successful entries
                                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()

View on GitHub (pinned to 6469eb736d)

Solutions

1. Retry the sink operation; this is an unexpected API response, not a user bug
2. Verify you are talking to a real/AWS-compatible Kinesis endpoint if using a custom `endpoint`
3. If reproducible with a mock service, fix the mock to return one entry per record; file a bug with the connector maintainers

// wrap sink writes; on this invariant error, fail and let checkpoint recovery replay
match sink.finish().await {
  Err(e) if e.to_string().contains("not match the response record count") => {
    warn!("Kinesis API contract violated; will retry from checkpoint");
    return Err(e); // surface to checkpointing/retry machinery
  }
  r => r,
}

When it happens

Trigger: During `finish`, `output.records.len() != record_count` for a successful PutRecords response while streaming the batch.

Common situations: Practically only seen with AWS-side anomalies, SDK deserialization quirks, or mocked/misbehaving Kinesis-compatible endpoints that return truncated results.

Related errors

• failed to send records. sent {} out of {}, last err: code: [
• should not fail because we have set `data` and `partition_ke
• 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/59eb9cbf652077aa. Report an issue: GitHub.