{"record":{"id":"59eb9cbf652077aa","repo":"risingwavelabs/risingwave","slug":"request-record-count-not-match-the-response-rec","errorCode":null,"errorMessage":"request record count {} not match the response record count {}","messagePattern":"request record count (.+?) not match the response record count (.+?)","errorType":"exception","errorClass":"SinkError::Kinesis","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/kinesis.rs","lineNumber":296,"sourceCode":"                    }\n                    if records.is_empty() {\n                        // at least include one record even if its size exceed `MAX_TOTAL_RECORD_PAYLOAD_SIZE`\n                        records.push(self.entries[start_idx].0.clone());\n                    }\n\n                    // 2. send the records and handle the result\n                    let record_count = records.len();\n                    match self\n                        .client\n                        .put_records()\n                        .stream_name(&self.stream_name)\n                        .set_records(Some(records))\n                        .send()\n                        .await\n                    {\n                        Ok(output) => {\n                            if record_count != output.records.len() {\n                                return Err(SinkError::Kinesis(anyhow!(\"request record count {} not match the response record count {}\", record_count, output.records.len())));\n                            }\n                            // From the doc of `put_records`:\n                            // A single record failure does not stop the processing of subsequent records. As a result,\n                            // PutRecords doesn’t guarantee the ordering of records. If you need to read records in the same\n                            // order they are written to the stream, use PutRecord instead of PutRecords, and write to the same shard.\n\n                            // Therefore, to ensure at least once and eventual consistency, we figure out the first failed entry, and retry\n                            // all the following entries even if the following entries may have been successfully processed.\n                            if let Some((first_failed_idx, result_entry)) = Self::first_failed_entry(output) {\n                                // first_failed_idx is also the number of successful entries\n                                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()","sourceCodeStart":278,"sourceCodeEnd":314,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/kinesis.rs#L278-L314","documentation":"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.","triggerScenarios":"During `finish`, `output.records.len() != record_count` for a successful PutRecords response while streaming the batch.","commonSituations":"Practically only seen with AWS-side anomalies, SDK deserialization quirks, or mocked/misbehaving Kinesis-compatible endpoints that return truncated results.","solutions":["Retry the sink operation; this is an unexpected API response, not a user bug","Verify you are talking to a real/AWS-compatible Kinesis endpoint if using a custom `endpoint`","If reproducible with a mock service, fix the mock to return one entry per record; file a bug with the connector maintainers"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"// wrap sink writes; on this invariant error, fail and let checkpoint recovery replay\nmatch sink.finish().await {\n  Err(e) if e.to_string().contains(\"not match the response record count\") => {\n    warn!(\"Kinesis API contract violated; will retry from checkpoint\");\n    return Err(e); // surface to checkpointing/retry machinery\n  }\n  r => r,\n}","preventionTips":["Use real AWS Kinesis or fully API-compatible endpoints","Report persistent occurrences as a bug with SDK/endpoint details","Keep checkpoints frequent so partial batches can be replayed safely"],"tags":["kinesis","invariant","put-records","api-response"],"backgroundTag":"unexpected-response-shape","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"}