redpanda-data / redpanda-data/connect
opensearch output: action does not support "create"
Nobody has claimed this yet.
- Dominant language
- Go
- Stars
- 8.8k
- Forks
- 969
- Avg merge
- 1d 13h
- Merged PRs (30d)
- 64
Description
Description
The output opensearch action does not support create, as a result can not sink messages into index of datastream type in opensearch
Location
The supportted action is in internal/impl/opensearch/output.go
Test
Preparation
Opensearch: 3.6.0
Kafka: 4.0.0
Redpandadata/connect: 4.90.3
Create events as a datastream in opensearch
Configure connector output
output_resources:
- label: o_kafka_os
broker:
pattern: greedy
copies: 5
outputs:
- drop_on:
error_patterns:
- illegal_argument_exception
- Compressor detection
- invalid json
output:
opensearch:
urls: [ "${OS_URI}" ]
action: index
index: 'events'
id: '${! uuid_v4() }'
Result
Input data in kafka, and then redpanda-connect will has following error in log:
time="2026-06-15T02:44:20Z" level=error msg="Failed to send message to opensearch: illegal_argument_exception: only write ops with an op_type of create are allowed in data streams" @service=redpanda-connect label="" path=root.output_resources.broker.outputs.0.drop_on.output
time="2026-06-15T02:44:20Z" level=warning msg="Message dropped due to error matching pattern 0: illegal_argument_exception: only write ops with an op_type of create are allowed in data streams" @service=redpanda-connect label="" path=root.output_resources.broker.outputs.0
Proposal
Add the following code into internal/impl/opensearch/output.go, just like elasticsearch_v9
case "index":
r = &opensearchutil.BulkIndexerItem{
Index: p.Index,
Action: "index",
Body: bytes.NewReader(p.Payload),
}
if p.ID != "" {
r.DocumentID = p.ID
}
if p.Routing != "" {
r.Routing = &p.Routing
}
+ case "create":
+ r = &opensearchutil.BulkIndexerItem{
+ Index: p.Index,
+ Action: "create",
+ Body: bytes.NewReader(p.Payload),
+ }
+ if p.ID != "" {
+ r.DocumentID = p.ID
+ }
+ if p.Routing != "" {
+ r.Routing = &p.Routing
+ }
Then can set the action to "create", then can write data into a index of datastream in opensearch
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start in internal/impl/opensearch/output.go and compare the existing action handling with the analogous elasticsearch_v9 implementation mentioned in the issue. Add support for configuring the OpenSearch action as "create", preserving the documented index, ID, and routing behavior. Done means messages can be written to an OpenSearch data stream without the op_type error.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- go
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 2/5
- Estimated time
- 1-3 hours
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 78/100