{"record":{"id":"8d082b32939fec92","repo":"risingwavelabs/risingwave","slug":"turbopuffer-sink-received-non-success-response","errorCode":null,"errorMessage":"Turbopuffer sink received non-success response: {} {}","messagePattern":"Turbopuffer sink received non-success response: (.+?) (.+?)","errorType":"http","errorClass":"SinkError::Http","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/turbopuffer.rs","lineNumber":678,"sourceCode":"        }))\n        .await?;\n        Ok(())\n    }\n}\n\nasync fn send_turbopuffer_request(client: reqwest::Client, url: String, body: Value) -> Result<()> {\n    let resp = client\n        .post(url)\n        .json(&body)\n        .send()\n        .await\n        .context(\"turbopuffer write request failed\")\n        .map_err(SinkError::Http)?;\n\n    if !resp.status().is_success() {\n        let status = resp.status();\n        let body = resp.text().await.unwrap_or_default();\n        return Err(SinkError::Http(anyhow!(\n            \"Turbopuffer sink received non-success response: {} {}\",\n            status,\n            body\n        )));\n    }\n    Ok(())\n}\n\n#[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize)]\n#[serde(untagged)]\nenum DocumentId {\n    U64(u64),\n    String(String),\n}\n\n#[derive(Debug)]\nenum CompactedOp {\n    Upsert(Map<String, Value>),","sourceCodeStart":660,"sourceCodeEnd":696,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/turbopuffer.rs#L660-L696","documentation":"send_turbopuffer_request performs the HTTP POST/DELETE to the Turbopuffer v2 API and requires a 2xx status. If the API returns any non-success status, the error includes the HTTP status code and the response body text. This surfaces auth failures, invalid payloads, quota issues, and Turbopuffer-side validation errors.","triggerScenarios":"flush_all sends a buffered write/delete payload and the Turbopuffer API responds with a non-2xx status (401 wrong API key, 400 malformed payload/schema mismatch, 429 rate limited, 5xx server error).","commonSituations":"Expired or wrong TURBOPUFFER_API_KEY; schema in the upsert not matching an existing namespace's schema; exceeding rate limits during heavy streaming writes; network/proxy returning error pages.","solutions":["Read the status and body in the error message to identify the cause (401/403 → fix the API key; 400 → fix schema/payload; 429 → reduce throughput).","Verify the API key and that the namespace/project are correct.","Check that column types in the sink match the Turbopuffer namespace schema; drop or recreate the namespace if the schema changed.","Retry with backoff for 429/5xx responses."],"exampleFix":"// before: expired key\nWITH (connector='turbopuffer', turbopuffer.api_key='sk-old', ...)\n// after: valid key via secret\nWITH (connector='turbopuffer', turbopuffer.api_key=secret turbopuffer_api_key, ...)","handlingStrategy":"retry","validationCode":"// before creating the sink, verify credentials with a direct API call:\n// curl -s -o /dev/null -w '%{http_code}' -H \"Authorization: Bearer $TPUF_KEY\" https://api.turbopuffer.com/v1/namespaces","typeGuard":null,"tryCatchPattern":"match res {\n    Ok(()) => {}\n    Err(e) if e.to_string().contains(\"non-success response: 429\") => retry_with_backoff(e),\n    Err(e) => return Err(e), // 4xx: do not retry, fix key/schema\n}","preventionTips":["Store the API key as a RisingWave secret, not inline.","Keep sink column types in sync with the Turbopuffer namespace schema.","Monitor rate limits; tune sink throughput/flush intervals for heavy streams."],"tags":["turbopuffer","http","sink","api"],"backgroundTag":"http-error-response","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}