feat(connectors): add Meilisearch source connector

Cargo.toml

@@ -39,12 +39,14 @@ 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",
     "core/connectors/sinks/stdout_sink",
     "core/connectors/sources/elasticsearch_source",
     "core/connectors/sources/influxdb_source",
+    "core/connectors/sources/meilisearch_source",
     "core/connectors/sources/postgres_source",
     "core/connectors/sources/random_source",
     "core/consensus",
@@ -194,6 +196,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

@@ -96,6 +96,7 @@ Please refer to the **[Source documentation](https://github.com/apache/iggy/tree
 ### Available Sources
 
 - **Elasticsearch Source** - polls documents from Elasticsearch indices
+- **Meilisearch Source** - polls documents from Meilisearch indices
 - **PostgreSQL Source** - reads rows from PostgreSQL tables with multiple consumption strategies (delete after read, mark as processed, timestamp tracking)
 - **Random Source** - generates random test messages (useful for testing/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,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_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
+
+[package.metadata.cargo-machete]
+ignored = ["dashmap", "once_cell"]
+
+[lib]
+crate-type = ["cdylib", "lib"]
+
+[dependencies]
+async-trait = { workspace = true }
+base64 = { workspace = true }
+dashmap = { workspace = true }
+iggy_common = { workspace = true }
+iggy_connector_sdk = { workspace = true }
+meilisearch-sdk = { workspace = true }
+once_cell = { workspace = true }
+reqwest = { workspace = true }
+secrecy = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+simd-json = { workspace = true }
+tokio = { workspace = true }
+tracing = { workspace = true }

core/connectors/sinks/meilisearch_sink/README.md

@@ -0,0 +1,45 @@
+# 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.
+- `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`.
+- `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 retry attempts. 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 retry attempts while opening the index. Defaults to `5`.
+
+## 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
+fail the batch instead of being silently dropped.
+
+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.
+
+`wait_for_tasks=false` only skips waiting for document indexing tasks during
+`consume()`. 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 = 100
+timeout = "30s"
+wait_for_tasks = true