feat(connectors): add Meilisearch sink connector

Cargo.toml

@@ -39,6 +39,7 @@ members = [
     "core/connectors/sinks/http_sink",
     "core/connectors/sinks/iceberg_sink",
     "core/connectors/sinks/influxdb_sink",
+    "core/connectors/sinks/meilisearch_sink",
     "core/connectors/sinks/mongodb_sink",
     "core/connectors/sinks/postgres_sink",
     "core/connectors/sinks/quickwit_sink",
@@ -200,6 +201,11 @@ lending-iterator = "0.1.7"
 libc = "0.2.186"
 log = "0.4.30"
 lz4_flex = "0.13.1"
+meilisearch-sdk = { version = "0.33.0", default-features = false, features = [
+    "reqwest",
+    "tls",
+    "jwt_rust_crypto",
+] }
 message_bus = { path = "core/message_bus" }
 metadata = { path = "core/metadata" }
 mimalloc = "0.1"

core/connectors/README.md

@@ -83,6 +83,7 @@ Each sink should have its own, custom configuration, which is passed along with
 - **Doris Sink** - loads JSON messages into Apache Doris tables via the Stream Load HTTP API
 - **Elasticsearch Sink** - sends messages to Elasticsearch indices
 - **Iceberg Sink** - writes data to Apache Iceberg tables via REST catalog
+- **Meilisearch Sink** - indexes messages in Meilisearch
 - **PostgreSQL Sink** - stores messages in PostgreSQL database tables
 - **Quickwit Sink** - indexes messages in Quickwit search engine
 - **Stdout Sink** - prints messages to standard output (useful for debugging/development)

core/connectors/sinks/README.md

@@ -12,6 +12,7 @@ Sink connectors are responsible for writing data from Iggy streams to external s
 | **elasticsearch_sink** | Sends messages to Elasticsearch indices for full-text search and analytics |
 | **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 |
 | **postgres_sink** | Stores messages in PostgreSQL database tables with configurable schemas |
 | **quickwit_sink** | Indexes messages in Quickwit search engine for log analytics |
 | **stdout_sink** | Prints messages to standard output (useful for debugging and development) |

core/connectors/sinks/meilisearch_sink/Cargo.toml

@@ -0,0 +1,47 @@
+# 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_meilisearch_sink"
+version = "0.4.0"
+description = "Iggy Meilisearch sink connector"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "search", "meilisearch"]
+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 }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+meilisearch-sdk = { 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 }

core/connectors/sinks/meilisearch_sink/README.md

@@ -0,0 +1,50 @@
+# Meilisearch Sink Connector
+
+A sink connector that consumes messages from Iggy streams and writes them to a
+Meilisearch index through the official Rust SDK.
+
+## Configuration
+
+- `url`: Meilisearch base URL. Paths, query strings, and fragments are ignored.
+- `index`: Target index UID.
+- `api_key`: Optional Meilisearch API key sent as `Authorization: Bearer`.
+- `primary_key`: Index primary key field. Defaults to `iggy_id`.
+- `document_action`: `replace` uses SDK add-or-replace semantics; `update` uses SDK add-or-update semantics. Defaults to `replace`.
+- `create_index_if_not_exists`: Create the index during `open()` when missing. Defaults to `true`.
+- `include_metadata`: Add Iggy metadata fields to each document. Defaults to `true`.
+- `batch_size`: Maximum documents per Meilisearch document request. Defaults to `1000`.
+- `timeout`: Request timeout as a humantime string, for example `30s`. Defaults to `30s`.
+- `wait_for_tasks`: Poll Meilisearch tasks until terminal state before returning from `consume()`. Defaults to `true`. Setting this to `false` makes document indexing fire-and-forget, so asynchronous Meilisearch task failures are not observed by the connector.
+- `task_timeout`: Maximum time to wait for each Meilisearch task. Defaults to `30s`.
+- `task_poll_interval`: Delay between task polls. Defaults to `100ms`.
+- `max_retries`: Maximum transient retries after the initial request. Defaults to `3`.
+- `retry_delay`: Initial transient retry delay. Defaults to `500ms`.
+- `max_retry_delay`: Maximum transient retry delay. Defaults to `5s`.
+- `max_open_retries`: Maximum transient retries after the initial request while opening the index. Defaults to `5`. This also applies to `get_task` polls while waiting for index creation during `open()`.
+
+## Behavior
+
+JSON object payloads are indexed as documents. JSON arrays or scalar values are
+wrapped in a `value` field because Meilisearch documents must be objects. Raw
+payloads are parsed as JSON when possible; otherwise, they are indexed as base64
+data. Text payloads are indexed in a `text` field. Unsupported payload schemas
+are skipped with a warning and counted as sink errors, matching the connector
+runtime's per-record drop behavior for malformed records.
+
+When the configured primary key is absent, the connector injects a stable value
+derived from the exact Iggy stream, topic, partition, offset, and message ID.
+This avoids Meilisearch primary-key inference failures and keeps repeated
+delivery idempotent for the same message.
+
+When `include_metadata` is enabled, metadata fields are only inserted when the
+document does not already contain those names. Existing user fields are
+preserved. If `primary_key` is set to a field other than `iggy_id`, the
+connector also inserts `iggy_id` as stable Iggy metadata when absent.
+
+`wait_for_tasks=false` only skips waiting for document indexing tasks during
+`consume()`. In that mode, successful submission lets the runtime commit the
+consumer offset before Meilisearch has confirmed indexing, so later task
+failures are not retried, logged, or counted by this connector. If
+`create_index_if_not_exists=true` and the connector creates the index during
+`open()`, it still waits for that index-creation task so the first batch cannot
+race the index creation.

core/connectors/sinks/meilisearch_sink/config.toml

@@ -0,0 +1,43 @@
+# 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 = "meilisearch"
+enabled = true
+version = 0
+name = "Meilisearch sink"
+path = "../../target/release/libiggy_connector_meilisearch_sink"
+verbose = false
+
+[[streams]]
+stream = "test_stream"
+topics = ["test_topic"]
+schema = "json"
+batch_length = 100
+poll_interval = "5ms"
+consumer_group = "meilisearch_sink_cg"
+
+[plugin_config]
+url = "http://localhost:7700"
+index = "iggy_messages"
+primary_key = "iggy_id"
+document_action = "replace"
+create_index_if_not_exists = true
+include_metadata = true
+batch_size = 1000
+timeout = "30s"
+wait_for_tasks = true