risingwavelabs/risingwave · error · SinkError::ElasticSearchOpenSearch

send bulk to elasticsearch failed: {:?}

Error message

send bulk to elasticsearch failed: {:?}

What it means

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.

Source

Thrown at src/connector/src/sink/elasticsearch_opensearch/elasticsearch_opensearch_client.rs:253

            };

            if bulks.len() >= self.config.batch_num_messages
                || bulks_size >= self.config.batch_size_kb * 1024
            {
                all_bulks.push(bulks);
                bulks = Vec::with_capacity(chunk_capacity);
                bulks_size = 0;
            }
        }
        if !bulks.is_empty() {
            all_bulks.push(bulks);
        }
        for bulks in all_bulks {
            let client_clone = self.client.clone();
            let future = async move {
                let result = client_clone.send(bulks).await?;
                if result["errors"].as_bool().is_none() || result["errors"].as_bool().unwrap() {
                    Err(SinkError::ElasticSearchOpenSearch(anyhow!(
                        "send bulk to elasticsearch failed: {:?}",
                        result
                    )))
                } else {
                    Ok(())
                }
            }
            .boxed();
            add_future.add_future_may_await(future).await?;
        }
        Ok(())
    }
}

View on GitHub (pinned to 6469eb736d)

Solutions

1. Inspect the wrapped `result` in the error message: items[].error explains the per-item failure.
2. Create the target index with the correct mapping (or enable auto-create) before writing.
3. Fix data-to-mapping type conflicts in the materialized view schema.
4. Verify the index name and credentials in the sink options.

Example fix

// before: index missing -> ensure mapping exists
PUT /my-index { "mappings": { "properties": { "user_id": { "type": "keyword" } } } }
// after: sink writes succeed
CREATE SINK es_sink FROM mv WITH (
  'connector' = 'elasticsearch',
  'url' = 'http://localhost:9200',
  'index' = 'my-index'
)

curl -s -XPUT 'http://localhost:9200/my-index' -H 'Content-Type: application/json' -d '{"mappings":{"properties":{"user_id":{"type":"keyword"}}}}'
curl -s 'http://localhost:9200/my-index/_mapping' | jq .   # confirm mapping matches MV column types

// the error message embeds the full bulk response; parse items[].error to find the failing doc
let resp: serde_json::Value = parse_from_error_message(msg);
for item in resp["items"].as_array().unwrap() { if let Some(e) = item["index"]["error"].as_object() { log::error!("bulk item failed: {:?}", e); } }

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "API error: {status}" and "HTTP 401/403/404/429/5xx" errors: non-2xx HTTP responses explained — this error's family across 27 libraries.

Related errors

• ElasticSearch/OpenSearch error: {0}
• Cannot find {}
• SinkError::ElasticSearchOpenSearch(anyhow!(err))
• Can't find data
• Insert error: {:?}, error url: {:?}

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/1f75065d7878517b. Report an issue: GitHub.