{"record":{"id":"9c92cbfa7263a0ae","repo":"risingwavelabs/risingwave","slug":"sinkerror-elasticsearchopensearch-anyhow-err","errorCode":null,"errorMessage":"SinkError::ElasticSearchOpenSearch(anyhow!(err))","messagePattern":"SinkError::ElasticSearchOpenSearch\\(anyhow!\\(err\\)\\)","errorType":"exception","errorClass":"SinkError::ElasticSearchOpenSearch","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/mod.rs","lineNumber":1297,"sourceCode":"        SinkError::Remote(anyhow!(value))\n    }\n}\n\nimpl From<RedisError> for SinkError {\n    fn from(value: RedisError) -> Self {\n        SinkError::Redis(value.to_report_string())\n    }\n}\n\nimpl From<tiberius::error::Error> for SinkError {\n    fn from(err: tiberius::error::Error) -> Self {\n        SinkError::SqlServer(anyhow!(err))\n    }\n}\n\nimpl From<::elasticsearch::Error> for SinkError {\n    fn from(err: ::elasticsearch::Error) -> Self {\n        SinkError::ElasticSearchOpenSearch(anyhow!(err))\n    }\n}\n\nimpl From<::opensearch::Error> for SinkError {\n    fn from(err: ::opensearch::Error) -> Self {\n        SinkError::ElasticSearchOpenSearch(anyhow!(err))\n    }\n}\n\nimpl From<tokio_postgres::Error> for SinkError {\n    fn from(err: tokio_postgres::Error) -> Self {\n        SinkError::Postgres(anyhow!(err))\n    }\n}\n\n#[cfg(test)]\nmod tests {\n    use std::collections::BTreeMap;","sourceCodeStart":1279,"sourceCodeEnd":1315,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/mod.rs#L1279-L1315","documentation":"RisingWave converts any ::elasticsearch::Error into the ElasticSearchOpenSearch SinkError variant. It means an Elasticsearch client operation (index, bulk, connection) failed and the driver error was wrapped into an anyhow error.","triggerScenarios":"Any elasticsearch crate error during sink writes: failed HTTP request, non-success Elasticsearch response, connection/transport failure, or deserialization failure of the response.","commonSituations":"Elasticsearch cluster unreachable or restarting, index mapping conflicts (wrong field type), bulk request too large, authentication against a secured cluster missing, or elasticsearch crate version mismatch with the cluster.","solutions":["Check cluster health and reachability: GET _cluster/health from the sink URL.","Inspect the wrapped HTTP status/body in the error message; fix mapping conflicts by updating the index mapping or coercing data.","Verify credentials/API key if security (X-Pack) is enabled on the cluster.","Reduce bulk batch size if 429/413 errors appear.","Align the elasticsearch crate major version with the cluster version."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// pre-check cluster availability before sink creation\ncurl -fsS -u \"$ES_USER:$ES_PASS\" \"$ELASTIC_URL/_cluster/health\" | jq -e '.status'\n","typeGuard":"fn is_retryable_es_error(err: &elasticsearch::Error) -> bool {\n    use elasticsearch::Error::*;\n    matches!(err, Io(_) | ServerError(_))\n}","tryCatchPattern":"// inspect the wrapped response status before deciding retry\nmatch err {\n    e if e.to_string().contains(\"429\") || e.to_string().contains(\"413\") => retry_with_backoff(),\n    _ => alert_and_fail(),\n}","preventionTips":["Pre-create the index with the correct mapping before starting the sink.","Use API keys/credentials configured via secrets, not inline DDL.","Cap bulk batch sizes below the cluster's http.max_content_length.","Monitor _cluster/health and catch throttling (429) early."],"tags":["rust","elasticsearch","sink","http"],"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-14T16:17:12.679Z"}