{"record":{"id":"1f75065d7878517b","repo":"risingwavelabs/risingwave","slug":"send-bulk-to-elasticsearch-failed","errorCode":null,"errorMessage":"send bulk to elasticsearch failed: {:?}","messagePattern":"send bulk to elasticsearch failed: (.+?)","errorType":"error_code","errorClass":"SinkError::ElasticSearchOpenSearch","httpStatus":null,"severity":"error","filePath":"src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_client.rs","lineNumber":253,"sourceCode":"            };\n\n            if bulks.len() >= self.config.batch_num_messages\n                || bulks_size >= self.config.batch_size_kb * 1024\n            {\n                all_bulks.push(bulks);\n                bulks = Vec::with_capacity(chunk_capacity);\n                bulks_size = 0;\n            }\n        }\n        if !bulks.is_empty() {\n            all_bulks.push(bulks);\n        }\n        for bulks in all_bulks {\n            let client_clone = self.client.clone();\n            let future = async move {\n                let result = client_clone.send(bulks).await?;\n                if result[\"errors\"].as_bool().is_none() || result[\"errors\"].as_bool().unwrap() {\n                    Err(SinkError::ElasticSearchOpenSearch(anyhow!(\n                        \"send bulk to elasticsearch failed: {:?}\",\n                        result\n                    )))\n                } else {\n                    Ok(())\n                }\n            }\n            .boxed();\n            add_future.add_future_may_await(future).await?;\n        }\n        Ok(())\n    }\n}\n","sourceCodeStart":235,"sourceCodeEnd":267,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_client.rs#L235-L267","documentation":"The bulk indexing HTTP response from Elasticsearch/OpenSearch either has no `errors` field or reports `errors: true`, meaning at least one bulk item failed. The client collects responses concurrently via futures and fails write_chunk whenever any bulk response signals item-level failures.","triggerScenarios":"write_chunk sends bulks to the ES/OpenSearch _bulk endpoint and a response body contains `\"errors\": true` (per-item failures such as mapping conflicts, index_not_found, version conflicts) or a malformed response missing the `errors` field.","commonSituations":"Target index missing and auto-creation disabled; dynamic mapping rejecting field types (e.g. sending a string to a date field); wrong index name casing; authentication returning an error-shaped body.","solutions":["Inspect the wrapped `result` in the error message: items[].error explains the per-item failure.","Create the target index with the correct mapping (or enable auto-create) before writing.","Fix data-to-mapping type conflicts in the materialized view schema.","Verify the index name and credentials in the sink options."],"exampleFix":"// before: index missing -> ensure mapping exists\nPUT /my-index { \"mappings\": { \"properties\": { \"user_id\": { \"type\": \"keyword\" } } } }\n// after: sink writes succeed\nCREATE SINK es_sink FROM mv WITH (\n  'connector' = 'elasticsearch',\n  'url' = 'http://localhost:9200',\n  'index' = 'my-index'\n)","handlingStrategy":"try-catch","validationCode":"curl -s -XPUT 'http://localhost:9200/my-index' -H 'Content-Type: application/json' -d '{\"mappings\":{\"properties\":{\"user_id\":{\"type\":\"keyword\"}}}}'\ncurl -s 'http://localhost:9200/my-index/_mapping' | jq .   # confirm mapping matches MV column types","typeGuard":null,"tryCatchPattern":"// the error message embeds the full bulk response; parse items[].error to find the failing doc\nlet resp: serde_json::Value = parse_from_error_message(msg);\nfor item in resp[\"items\"].as_array().unwrap() { if let Some(e) = item[\"index\"][\"error\"].as_object() { log::error!(\"bulk item failed: {:?}\", e); } }","preventionTips":["Create the index with an explicit mapping before starting the sink.","Ensure MV column types match the ES mapping (keyword vs text vs date).","Keep index names lowercase — ES rejects uppercase names at write time."],"tags":["elasticsearch","opensearch","sink","bulk","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"}