feat(connectors): add OpenSearch sink connector

.config/nextest.toml

@@ -48,6 +48,20 @@ max-threads = 1
 filter = 'package(integration) and test(/connectors::elasticsearch::/)'
 test-group = "elasticsearch"
 
+# OpenSearch tests share one reusable container (fixed name
+# `iggy-test-opensearch`, ReuseDirective::Always), same reasoning as
+# elasticsearch above. Serializing the group lets the first test create it and
+# the rest attach by name, instead of racing to create the same name
+# concurrently (Docker 409 Conflict) or hammering a single freshly-started
+# instance with concurrent requests before it has stabilized. Per-test
+# isolation comes from a unique index per fixture, not a fresh container.
+[test-groups.opensearch]
+max-threads = 1
+
+[[profile.default.overrides]]
+filter = 'package(integration) and test(/connectors::opensearch::/)'
+test-group = "opensearch"
+
 [profile.default]
 slow-timeout = { period = "60s", terminate-after = 5 }
 

Cargo.toml

@@ -42,6 +42,7 @@ members = [
     "core/connectors/sinks/influxdb_sink",
     "core/connectors/sinks/meilisearch_sink",
     "core/connectors/sinks/mongodb_sink",
+    "core/connectors/sinks/opensearch_sink",
     "core/connectors/sinks/postgres_sink",
     "core/connectors/sinks/quickwit_sink",
     "core/connectors/sinks/s3_sink",
@@ -234,6 +235,7 @@ nix = { version = "0.31.3", features = ["feature", "fs", "resource", "sched"] }
 nonzero_lit = "0.1.2"
 notify = "8.2.0"
 octocrab = "0.54.0"
+opensearch = { version = "2.4.0", default-features = false, features = ["rustls-tls"] }
 opentelemetry = { version = "0.32.0", features = ["trace", "logs"] }
 opentelemetry-appender-tracing = { version = "0.32.0", features = ["log"] }
 opentelemetry-otlp = { version = "0.32.0", features = [

core/connectors/README.md

@@ -84,6 +84,7 @@ Each sink should have its own, custom configuration, which is passed along with
 - **Elasticsearch Sink** - sends messages to Elasticsearch indices
 - **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog
 - **Meilisearch Sink** - indexes messages in Meilisearch
+- **OpenSearch Sink** - indexes messages in OpenSearch for full-text search
 - **PostgreSQL Sink** - stores messages in PostgreSQL database tables
 - **Quickwit Sink** - indexes messages in Quickwit search engine
 - **S3 Sink** - writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces)

core/connectors/runtime/example_config/connectors/opensearch_sink.toml

@@ -0,0 +1,49 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+type = "sink"
+key = "opensearch"
+enabled = true
+version = 0
+name = "OpenSearch sink"
+path = "target/release/libiggy_connector_opensearch_sink"
+verbose = false
+
+[[streams]]
+stream = "example_stream"
+topics = ["example_topic"]
+schema = "json"
+batch_length = 1000
+poll_interval = "5ms"
+consumer_group = "opensearch_sink_connector"
+
+[plugin_config]
+url = "http://localhost:9200"
+index = "iggy_messages"
+# username = "admin"
+# password = "..."
+# document_id_field = "order_id"
+create_index_if_not_exists = true
+include_metadata = true
+batch_size = 1000
+timeout = "30s"
+refresh = "false"
+max_retries = 3
+retry_delay = "500ms"
+max_retry_delay = "5s"
+max_open_retries = 5
+verbose_logging = false

core/connectors/sinks/README.md

@@ -13,6 +13,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s
 | **iceberg_sink** | Writes data to Apache Iceberg tables via REST catalog with S3/GCS/Azure storage |
 | **influxdb_sink** | Writes messages to InfluxDB as line-protocol points; supports both V2 (org/bucket, Flux) and V3 (db, SQL) |
 | **meilisearch_sink** | Indexes messages in Meilisearch for full-text search |
+| **opensearch_sink** | Indexes messages in OpenSearch for full-text search and retrieval |
 | **postgres_sink** | Stores messages in PostgreSQL database tables with configurable schemas |
 | **quickwit_sink** | Indexes messages in Quickwit search engine for log analytics |
 | **s3_sink** | Writes messages to Amazon S3 and S3-compatible stores (MinIO, R2, B2, DO Spaces) |

core/connectors/sinks/opensearch_sink/Cargo.toml

@@ -0,0 +1,52 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "iggy_connector_opensearch_sink"
+version = "0.5.0-edge.1"
+description = "Iggy OpenSearch sink connector"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "opensearch", "sink"]
+categories = ["command-line-utilities", "database", "network-programming"]
+homepage = "https://iggy.apache.org"
+documentation = "https://iggy.apache.org/docs"
+repository = "https://github.com/apache/iggy"
+readme = "../../README.md"
+publish = false
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }
+bytes = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+opensearch = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }
+url = { workspace = true }
+
+[dev-dependencies]
+toml = { workspace = true }
+wiremock = { workspace = true }