This is an automated email from the ASF dual-hosted git repository.
slbotbm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 3ca8695d3 feat(connectors): add Meilisearch sink connector (#3497)
3ca8695d3 is described below
commit 3ca8695d330e17c7eaa3f269c19cb1ee90354100
Author: Diaconu Radu-Mihai <[email protected]>
AuthorDate: Tue Aug 4 02:08:16 2026 +0300
feat(connectors): add Meilisearch sink connector (#3497)
## Summary
- add the Meilisearch sink connector using the official
`meilisearch-sdk`
- document the sink configuration and payload mapping behavior
- add Docker-backed integration coverage for indexing JSON messages into
Meilisearch
## Motivation
This splits the sink portion out of #3404 so the Meilisearch sink can be
reviewed and merged independently from the source connector.
Closes #3495.
## Validation
- `cargo sort --no-format --workspace`
- `cargo test -p iggy_connector_meilisearch_sink`
- `cargo clippy -p iggy_connector_meilisearch_sink --all-targets -- -D
warnings`
- `env CARGO_BIN_EXE_iggy-server=... CARGO_BIN_EXE_iggy-connectors=...
cargo test -p integration --test mod --
connectors::meilisearch::meilisearch_sink`
- `git diff --check`
---------
Co-authored-by: radudiaconu <[email protected]>
Co-authored-by: Hubert Gruszecki <[email protected]>
Co-authored-by: Rimuksh Kansal <[email protected]>
---
Cargo.lock | 90 ++
Cargo.toml | 6 +
core/connectors/README.md | 1 +
core/connectors/sinks/README.md | 1 +
core/connectors/sinks/meilisearch_sink/Cargo.toml | 47 +
core/connectors/sinks/meilisearch_sink/README.md | 105 ++
core/connectors/sinks/meilisearch_sink/config.toml | 43 +
core/connectors/sinks/meilisearch_sink/src/lib.rs | 1412 ++++++++++++++++++++
.../connectors/fixtures/meilisearch/container.rs | 180 +++
.../tests/connectors/fixtures/meilisearch/mod.rs | 22 +
.../tests/connectors/fixtures/meilisearch/sink.rs | 90 ++
core/integration/tests/connectors/fixtures/mod.rs | 2 +
.../connectors/meilisearch/meilisearch_sink.rs | 81 ++
.../tests/connectors/meilisearch/mod.rs | 18 +
.../tests/connectors/meilisearch/sink.toml | 20 +
core/integration/tests/connectors/mod.rs | 1 +
16 files changed, 2119 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index 0ecf6332b..3cd7dccbc 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3300,6 +3300,15 @@ dependencies = [
"unicode-segmentation",
]
+[[package]]
+name = "convert_case"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "baaaa0ecca5b51987b9423ccdc971514dd8b0bb7b4060b983d3664dad3f1f89f"
+dependencies = [
+ "unicode-segmentation",
+]
+
[[package]]
name = "convert_case"
version = "0.9.0"
@@ -7048,6 +7057,24 @@ dependencies = [
"uuid",
]
+[[package]]
+name = "iggy_connector_meilisearch_sink"
+version = "0.4.1-edge.1"
+dependencies = [
+ "async-trait",
+ "base64",
+ "iggy_common",
+ "iggy_connector_sdk",
+ "meilisearch-sdk",
+ "secrecy",
+ "serde",
+ "serde_json",
+ "simd-json",
+ "tokio",
+ "tracing",
+ "url",
+]
+
[[package]]
name = "iggy_connector_mongodb_sink"
version = "0.4.1-edge.1"
@@ -7536,6 +7563,15 @@ version = "1.70.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695"
+[[package]]
+name = "iso8601"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "74a0559b45528cf0732d911524974977a5749f477d7dd99652830ffdaf53c4d1"
+dependencies = [
+ "nom 8.0.0",
+]
+
[[package]]
name = "itertools"
version = "0.13.0"
@@ -8335,6 +8371,49 @@ version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "7ebb8d8732c6a6df3d8f032a82911cfc747e00efb95cc46e8d0acd5b5b88570c"
+[[package]]
+name = "meilisearch-index-setting-macro"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "93b5b21df781c820a9cc387b808d4128cbc164dd28d67ac6ed666a00996f8f15"
+dependencies = [
+ "convert_case 0.8.0",
+ "proc-macro2",
+ "quote",
+ "structmeta",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "meilisearch-sdk"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "19e6e3646ba2a9a306296c1edf4a050508a408c1b59ca456d9ad4965ec6e91e9"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "either",
+ "futures-channel",
+ "futures-core",
+ "futures-io",
+ "futures-util",
+ "iso8601",
+ "jsonwebtoken",
+ "log",
+ "meilisearch-index-setting-macro",
+ "pin-project-lite",
+ "reqwest 0.12.28",
+ "serde",
+ "serde_json",
+ "thiserror 2.0.19",
+ "time",
+ "tokio",
+ "uuid",
+ "wasm-bindgen-futures",
+ "web-sys",
+ "yaup",
+]
+
[[package]]
name = "memchr"
version = "2.8.3"
@@ -14971,6 +15050,17 @@ dependencies = [
"time",
]
+[[package]]
+name = "yaup"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b0144f1a16a199846cb21024da74edd930b43443463292f536b7110b4855b5c6"
+dependencies = [
+ "form_urlencoded",
+ "serde",
+ "thiserror 1.0.69",
+]
+
[[package]]
name = "yew"
version = "0.23.0"
diff --git a/Cargo.toml b/Cargo.toml
index 0147c62ed..0bd07364f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -40,6 +40,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",
@@ -216,6 +217,11 @@ lending-iterator = "0.1.7"
libc = "0.2.188"
log = "0.4.33"
lz4_flex = "0.14.0"
+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"
diff --git a/core/connectors/README.md b/core/connectors/README.md
index 64bfc9a1f..2690a9775 100644
--- a/core/connectors/README.md
+++ b/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
- **S3 Sink** - writes messages to Amazon S3 and S3-compatible stores (MinIO,
R2, B2, DO Spaces)
diff --git a/core/connectors/sinks/README.md b/core/connectors/sinks/README.md
index 57ea05549..e23e1ace9 100644
--- a/core/connectors/sinks/README.md
+++ b/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 |
| **s3_sink** | Writes messages to Amazon S3 and S3-compatible stores (MinIO,
R2, B2, DO Spaces) |
diff --git a/core/connectors/sinks/meilisearch_sink/Cargo.toml
b/core/connectors/sinks/meilisearch_sink/Cargo.toml
new file mode 100644
index 000000000..44233a9dc
--- /dev/null
+++ b/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.1-edge.1"
+description = "Iggy Meilisearch sink connector"
+edition = "2024"
+license = "Apache-2.0"
+keywords = ["iggy", "messaging", "streaming", "meilisearch", "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 }
+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 }
diff --git a/core/connectors/sinks/meilisearch_sink/README.md
b/core/connectors/sinks/meilisearch_sink/README.md
new file mode 100644
index 000000000..f164c83dd
--- /dev/null
+++ b/core/connectors/sinks/meilisearch_sink/README.md
@@ -0,0 +1,105 @@
+# 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
+
+```toml
+[plugin_config]
+url = "https://meilisearch.example.com";
+index = "iggy_messages"
+# api_key = "..."
+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
+task_timeout = "30s"
+task_poll_interval = "100ms"
+max_retries = 3
+retry_delay = "500ms"
+max_retry_delay = "5s"
+max_open_retries = 5
+```
+
+- `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`.
+ Use HTTPS for non-local Meilisearch hosts; HTTP sends the key without
+ transport encryption.
+- `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()`. Each retried open operation
+ uses the configured request `timeout` plus backoff; there is no single total
+ open deadline.
+
+## 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. Because the sink
+returns success after dropping an unsupported-schema record, the runtime can
+commit the consumer offset for that record. There is no built-in dead-letter
+queue for these drops.
+
+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. If the payload already
+contains the configured primary key, that value is preserved. Operators must
+ensure user-provided primary keys are unique, otherwise Meilisearch
+add-or-replace semantics can collapse distinct messages into one document.
+
+When `include_metadata` is enabled, the connector writes reserved `iggy_*`
+provenance fields after payload parsing. These fields overwrite same-named
+payload fields so audit metadata reflects the actual stream, topic, partition,
+offset, checksum, and timestamps. `iggy_checksum` is stored as a string to
avoid
+JSON number precision loss in Meilisearch clients. Offset and timestamp
metadata
+remain JSON numbers. If `primary_key` is set to a field other than `iggy_id`,
+the connector also writes `iggy_id` as stable Iggy metadata.
+
+## Delivery Semantics
+
+The connector runtime invokes `consume()` through an FFI callback whose status
+code is not currently used to gate offset commits. A batch error returned by
the
+sink is logged by the sink, but the runtime does not redeliver that batch. The
+effective runtime-level delivery guarantee is at-most-once on sink errors. The
+sink's retry settings only provide best-effort retries inside a single
+`consume()` call.
+
+`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. This mode is fire-and-forget and does not provide
+durability.
+
+The close-time counters are attempt counters. `documents_enqueued` counts
+documents accepted by completed Meilisearch SDK calls in this process, and
+`documents_confirmed` counts those same documents only when task waiting is
+enabled and the corresponding task reached success. `errors` includes invalid
+records plus documents in failed chunks and trailing chunks that were not
+attempted after an earlier chunk failed.
diff --git a/core/connectors/sinks/meilisearch_sink/config.toml
b/core/connectors/sinks/meilisearch_sink/config.toml
new file mode 100644
index 000000000..1d9963115
--- /dev/null
+++ b/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"
+
+[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
diff --git a/core/connectors/sinks/meilisearch_sink/src/lib.rs
b/core/connectors/sinks/meilisearch_sink/src/lib.rs
new file mode 100644
index 000000000..64b42ab8c
--- /dev/null
+++ b/core/connectors/sinks/meilisearch_sink/src/lib.rs
@@ -0,0 +1,1412 @@
+// 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.
+
+use async_trait::async_trait;
+use base64::{Engine as _, engine::general_purpose};
+use iggy_common::IggyTimestamp;
+use iggy_connector_sdk::{
+ ConsumedMessage, Error, MessagesMetadata, Payload, Sink, TopicMetadata,
+ retry::{exponential_backoff, jitter, parse_duration},
+ sink_connector,
+};
+use meilisearch_sdk::{
+ client::Client,
+ errors::{
+ Error as MeilisearchSdkError, ErrorCode as MeilisearchErrorCode,
+ ErrorType as MeilisearchErrorType,
+ },
+ indexes::Index,
+ task_info::TaskInfo,
+ tasks::Task,
+};
+use secrecy::{ExposeSecret, SecretString};
+use serde::{Deserialize, Serialize};
+use serde_json::{Map, Value, json};
+use std::{cmp, future::Future, net::IpAddr, time::Duration};
+use tokio::{
+ sync::Mutex,
+ time::{Instant, sleep},
+};
+use tracing::{debug, error, info, warn};
+use url::Url;
+
+sink_connector!(MeilisearchSink);
+
+const DEFAULT_PRIMARY_KEY: &str = "iggy_id";
+const DEFAULT_CREATE_INDEX_IF_NOT_EXISTS: bool = true;
+const DEFAULT_INCLUDE_METADATA: bool = true;
+const DEFAULT_BATCH_SIZE: usize = 1000;
+const DEFAULT_TIMEOUT: &str = "30s";
+const DEFAULT_WAIT_FOR_TASKS: bool = true;
+const DEFAULT_TASK_TIMEOUT: &str = "30s";
+const DEFAULT_TASK_POLL_INTERVAL: &str = "100ms";
+const DEFAULT_RETRY_DELAY: &str = "500ms";
+const DEFAULT_MAX_RETRY_DELAY: &str = "5s";
+const DEFAULT_MAX_RETRIES: u32 = 3;
+const DEFAULT_MAX_OPEN_RETRIES: u32 = 5;
+const ENCODING_BASE64: &str = "base64";
+
+#[derive(Debug)]
+struct State {
+ invocations_count: usize,
+ documents_enqueued: usize,
+ documents_confirmed: usize,
+ errors_count: usize,
+}
+
+#[derive(Debug, Default, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum MeilisearchDocumentAction {
+ #[default]
+ Replace,
+ Update,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct MeilisearchSinkConfig {
+ pub url: String,
+ pub index: String,
+ #[serde(serialize_with =
"iggy_common::serde_secret::serialize_optional_secret")]
+ pub api_key: Option<SecretString>,
+ pub primary_key: Option<String>,
+ pub document_action: Option<MeilisearchDocumentAction>,
+ pub create_index_if_not_exists: Option<bool>,
+ pub include_metadata: Option<bool>,
+ pub batch_size: Option<usize>,
+ pub timeout: Option<String>,
+ pub wait_for_tasks: Option<bool>,
+ pub task_timeout: Option<String>,
+ pub task_poll_interval: Option<String>,
+ pub max_retries: Option<u32>,
+ pub retry_delay: Option<String>,
+ pub max_retry_delay: Option<String>,
+ pub max_open_retries: Option<u32>,
+}
+
+#[derive(Debug)]
+pub struct MeilisearchSink {
+ id: u32,
+ config: ResolvedMeilisearchSinkConfig,
+ client: Option<Client>,
+ state: Mutex<State>,
+}
+
+#[derive(Debug)]
+struct ResolvedMeilisearchSinkConfig {
+ url: String,
+ index: String,
+ api_key: Option<SecretString>,
+ primary_key: String,
+ document_action: MeilisearchDocumentAction,
+ create_index_if_not_exists: bool,
+ include_metadata: bool,
+ batch_size: usize,
+ timeout: Duration,
+ wait_for_tasks: bool,
+ task_timeout: Duration,
+ task_poll_interval: Duration,
+ max_retries: u32,
+ retry_delay: Duration,
+ max_retry_delay: Duration,
+ max_open_retries: u32,
+}
+
+impl From<MeilisearchSinkConfig> for ResolvedMeilisearchSinkConfig {
+ fn from(config: MeilisearchSinkConfig) -> Self {
+ let primary_key = config
+ .primary_key
+ .map(|value| value.trim().to_string())
+ .filter(|value| !value.is_empty())
+ .unwrap_or_else(|| DEFAULT_PRIMARY_KEY.to_string());
+ let document_action = config.document_action.unwrap_or_default();
+ let create_index_if_not_exists = config
+ .create_index_if_not_exists
+ .unwrap_or(DEFAULT_CREATE_INDEX_IF_NOT_EXISTS);
+ let include_metadata =
config.include_metadata.unwrap_or(DEFAULT_INCLUDE_METADATA);
+ let batch_size =
config.batch_size.unwrap_or(DEFAULT_BATCH_SIZE).max(1);
+ let timeout = parse_duration(config.timeout.as_deref(),
DEFAULT_TIMEOUT);
+ let wait_for_tasks =
config.wait_for_tasks.unwrap_or(DEFAULT_WAIT_FOR_TASKS);
+ let task_timeout = parse_duration(config.task_timeout.as_deref(),
DEFAULT_TASK_TIMEOUT);
+ let task_poll_interval = parse_duration(
+ config.task_poll_interval.as_deref(),
+ DEFAULT_TASK_POLL_INTERVAL,
+ );
+ let max_retries = config.max_retries.unwrap_or(DEFAULT_MAX_RETRIES);
+ let mut retry_delay = parse_duration(config.retry_delay.as_deref(),
DEFAULT_RETRY_DELAY);
+ let mut max_retry_delay =
+ parse_duration(config.max_retry_delay.as_deref(),
DEFAULT_MAX_RETRY_DELAY);
+ if retry_delay > max_retry_delay {
+ warn!(
+ "Meilisearch sink retry_delay ({:?}) exceeds max_retry_delay
({:?}). Swapping values.",
+ retry_delay, max_retry_delay
+ );
+ std::mem::swap(&mut retry_delay, &mut max_retry_delay);
+ }
+ let max_open_retries =
config.max_open_retries.unwrap_or(DEFAULT_MAX_OPEN_RETRIES);
+
+ Self {
+ url: config.url,
+ index: config.index.trim().to_string(),
+ api_key: config.api_key,
+ primary_key,
+ document_action,
+ create_index_if_not_exists,
+ include_metadata,
+ batch_size,
+ timeout,
+ wait_for_tasks,
+ task_timeout,
+ task_poll_interval,
+ max_retries,
+ retry_delay,
+ max_retry_delay,
+ max_open_retries,
+ }
+ }
+}
+
+impl MeilisearchSink {
+ pub fn new(id: u32, config: MeilisearchSinkConfig) -> Self {
+ Self {
+ id,
+ config: config.into(),
+ client: None,
+ state: Mutex::new(State {
+ invocations_count: 0,
+ documents_enqueued: 0,
+ documents_confirmed: 0,
+ errors_count: 0,
+ }),
+ }
+ }
+
+ fn create_client(&self) -> Result<Client, Error> {
+ let url = normalize_host(&self.config.url)?;
+ warn_if_api_key_uses_insecure_http(&self.config.url, &url,
self.config.api_key.is_some());
+ let api_key = self.config.api_key.as_ref().map(|key|
key.expose_secret());
+ Client::new(url, api_key).map_err(|error| {
+ Error::Connection(format!("Failed to create Meilisearch client:
{error}"))
+ })
+ }
+
+ fn validate_config(&self) -> Result<(), Error> {
+ if self.config.index.is_empty() {
+ return Err(Error::InvalidConfigValue(
+ "Meilisearch index cannot be empty".to_string(),
+ ));
+ }
+
+ Ok(())
+ }
+
+ async fn check_connectivity(&self, client: &Client) -> Result<(), Error> {
+ let mut retries = 0u32;
+
+ loop {
+ let result = tokio::time::timeout(self.config.timeout,
client.health()).await;
+ match result {
+ Ok(Ok(health)) if health.status == "available" => return
Ok(()),
+ Ok(Ok(health)) => {
+ if retries >= self.config.max_open_retries {
+ return Err(Error::Connection(format!(
+ "Meilisearch health check returned status '{}'",
+ health.status
+ )));
+ }
+ retries += 1;
+ let delay = jitter(exponential_backoff(
+ self.config.retry_delay,
+ retries,
+ self.config.max_retry_delay,
+ ));
+ warn!(
+ "Meilisearch health check returned status '{}' (retry
{}/{}). Retrying in {:?}...",
+ health.status, retries, self.config.max_open_retries,
delay
+ );
+ sleep(delay).await;
+ }
+ Ok(Err(error)) => {
+ let should_retry =
+ retries < self.config.max_open_retries &&
is_transient_sdk_error(&error);
+ if !should_retry {
+ return Err(map_sdk_error(error));
+ }
+ retries += 1;
+ let delay = jitter(exponential_backoff(
+ self.config.retry_delay,
+ retries,
+ self.config.max_retry_delay,
+ ));
+ warn!(
+ "Meilisearch health check failed (retry {}/{}): {}.
Retrying in {:?}...",
+ retries, self.config.max_open_retries, error, delay
+ );
+ sleep(delay).await;
+ }
+ Err(_) => {
+ if retries >= self.config.max_open_retries {
+ return Err(Error::HttpRequestFailed(format!(
+ "Meilisearch health check timed out after {:?}",
+ self.config.timeout
+ )));
+ }
+ retries += 1;
+ let delay = jitter(exponential_backoff(
+ self.config.retry_delay,
+ retries,
+ self.config.max_retry_delay,
+ ));
+ warn!(
+ "Meilisearch health check timed out after {:?} (retry
{}/{}). Retrying in {:?}...",
+ self.config.timeout, retries,
self.config.max_open_retries, delay
+ );
+ sleep(delay).await;
+ }
+ }
+ }
+ }
+
+ async fn ensure_index_exists(&self, client: &Client) -> Result<(), Error> {
+ match self.get_index_if_exists(client).await? {
+ Some(index) => {
+ info!("Meilisearch index '{}' already exists",
self.config.index);
+ if let Some(primary_key) = index.primary_key.as_deref()
+ && primary_key != self.config.primary_key
+ {
+ warn!(
+ "Meilisearch index '{}' primary key '{}' differs from
configured primary key '{}'",
+ self.config.index, primary_key, self.config.primary_key
+ );
+ } else if index.primary_key.is_none() {
+ warn!(
+ "Meilisearch index '{}' does not currently have a
primary key. Configured primary key '{}' will be sent with document indexing
requests.",
+ self.config.index, self.config.primary_key
+ );
+ }
+ Ok(())
+ }
+ None if self.config.create_index_if_not_exists =>
self.create_index(client).await,
+ None => Err(Error::InitError(format!(
+ "Meilisearch index '{}' does not exist and
create_index_if_not_exists=false",
+ self.config.index
+ ))),
+ }
+ }
+
+ async fn get_index_if_exists(&self, client: &Client) ->
Result<Option<Index>, Error> {
+ self.retry_sdk_open_operation("get index", || async {
+ match client.get_index(&self.config.index).await {
+ Ok(index) => Ok(Some(index)),
+ Err(error) if is_index_not_found(&error) => Ok(None),
+ Err(error) => Err(error),
+ }
+ })
+ .await
+ }
+
+ async fn create_index(&self, client: &Client) -> Result<(), Error> {
+ info!(
+ "Creating Meilisearch index '{}' with primary key '{}'",
+ self.config.index, self.config.primary_key
+ );
+
+ let task = self
+ .retry_sdk_open_operation("create index", || {
+ client.create_index(&self.config.index,
Some(&self.config.primary_key))
+ })
+ .await?;
+ self.wait_for_index_creation_task(client, task).await?;
+
+ info!("Created Meilisearch index '{}'", self.config.index);
+ Ok(())
+ }
+
+ fn prepare_document(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ message: ConsumedMessage,
+ ) -> Result<Value, Error> {
+ let ConsumedMessage {
+ id: message_id,
+ offset,
+ checksum,
+ timestamp,
+ origin_timestamp,
+ headers,
+ payload,
+ } = message;
+
+ let mut document = match payload {
+ Payload::Json(value) => {
+
Self::document_from_json_value(owned_value_into_serde_json(value))
+ }
+ Payload::Raw(bytes) => {
+ let mut bytes_copy = bytes.clone();
+ match simd_json::from_slice::<simd_json::OwnedValue>(&mut
bytes_copy) {
+ Ok(value) =>
Self::document_from_json_value(owned_value_into_serde_json(value)),
+ Err(_) => Map::from_iter([
+ (
+ "data".to_string(),
+
Value::String(general_purpose::STANDARD.encode(&bytes)),
+ ),
+ ("data_type".to_string(),
Value::String("raw".to_string())),
+ (
+ "data_encoding".to_string(),
+ Value::String(ENCODING_BASE64.to_string()),
+ ),
+ ]),
+ }
+ }
+ Payload::Text(text) => Map::from_iter([
+ ("text".to_string(), Value::String(text)),
+ ("data_type".to_string(), Value::String("text".to_string())),
+ ]),
+ _ => {
+ return Err(Error::InvalidRecordValue(format!(
+ "Unsupported payload format for Meilisearch sink: {}",
+ messages_metadata.schema
+ )));
+ }
+ };
+
+ let mut generated_id = None;
+ if !document.contains_key(self.config.primary_key.as_str()) {
+ let value = generated_document_id_from_parts(
+ topic_metadata,
+ messages_metadata,
+ offset,
+ message_id,
+ )?;
+ if self.config.primary_key != DEFAULT_PRIMARY_KEY {
+ generated_id = Some(value.clone());
+ }
+ document.insert(self.config.primary_key.clone(),
Value::String(value));
+ }
+
+ if self.config.include_metadata {
+ if self.config.primary_key != DEFAULT_PRIMARY_KEY {
+ let id = match generated_id {
+ Some(id) => id,
+ None => generated_document_id_from_parts(
+ topic_metadata,
+ messages_metadata,
+ offset,
+ message_id,
+ )?,
+ };
+ upsert_metadata_field(&mut document, DEFAULT_PRIMARY_KEY,
Value::String(id));
+ }
+ upsert_metadata_field(
+ &mut document,
+ "iggy_message_id",
+ Value::String(message_id.to_string()),
+ );
+ upsert_metadata_field(&mut document, "iggy_offset",
Value::from(offset));
+ upsert_metadata_field(
+ &mut document,
+ "iggy_stream",
+ Value::from(topic_metadata.stream.as_str()),
+ );
+ upsert_metadata_field(
+ &mut document,
+ "iggy_topic",
+ Value::from(topic_metadata.topic.as_str()),
+ );
+ upsert_metadata_field(
+ &mut document,
+ "iggy_partition",
+ Value::from(messages_metadata.partition_id),
+ );
+ upsert_metadata_field(
+ &mut document,
+ "iggy_checksum",
+ Value::String(checksum.to_string()),
+ );
+ upsert_metadata_field(&mut document, "iggy_timestamp",
Value::from(timestamp));
+ upsert_metadata_field(
+ &mut document,
+ "iggy_origin_timestamp",
+ Value::from(origin_timestamp),
+ );
+ upsert_metadata_field(
+ &mut document,
+ "iggy_ingested_at",
+ Value::from(IggyTimestamp::now().as_millis() as i64),
+ );
+ if let Some(headers) = &headers
+ && let Ok(headers_value) = serde_json::to_value(headers)
+ {
+ upsert_metadata_field(&mut document, "iggy_headers",
headers_value);
+ }
+ }
+
+ Ok(Value::Object(document))
+ }
+
+ fn document_from_json_value(value: Value) -> Map<String, Value> {
+ match value {
+ Value::Object(object) => object,
+ other => {
+ let mut object = Map::new();
+ object.insert("value".to_string(), other);
+ object
+ }
+ }
+ }
+
+ async fn index_documents(
+ &self,
+ client: &Client,
+ documents: Vec<Value>,
+ ) -> Result<usize, PartialIndexError> {
+ let mut accepted = 0usize;
+ let mut accounted = 0usize;
+ let documents_count = documents.len();
+ for chunk in documents.chunks(self.config.batch_size) {
+ match self.index_document_chunk(client, chunk).await {
+ Ok(indexed) => {
+ accepted += indexed;
+ accounted += chunk.len();
+ }
+ Err(partial_error) => {
+ let accepted = accepted + partial_error.accepted;
+ let accounted = accounted + partial_error.accepted +
partial_error.failed;
+ let failed = partial_error.failed +
documents_count.saturating_sub(accounted);
+ return Err(PartialIndexError {
+ accepted,
+ failed,
+ error: partial_error.error,
+ });
+ }
+ }
+ }
+ Ok(accepted)
+ }
+
+ async fn index_document_chunk(
+ &self,
+ client: &Client,
+ documents: &[Value],
+ ) -> Result<usize, PartialIndexError> {
+ if documents.is_empty() {
+ return Ok(0);
+ }
+
+ let index = client.index(&self.config.index);
+ let task = match self.config.document_action {
+ MeilisearchDocumentAction::Replace => {
+ self.retry_sdk_operation("add or replace documents", || {
+ index.add_or_replace(documents,
Some(&self.config.primary_key))
+ })
+ .await
+ }
+ MeilisearchDocumentAction::Update => {
+ self.retry_sdk_operation("add or update documents", || {
+ index.add_or_update(documents,
Some(&self.config.primary_key))
+ })
+ .await
+ }
+ }
+ .map_err(|error| PartialIndexError {
+ accepted: 0,
+ failed: documents.len(),
+ error,
+ })?;
+ self.wait_for_task(client, task)
+ .await
+ .map_err(|error| PartialIndexError {
+ accepted: 0,
+ failed: documents.len(),
+ error,
+ })?;
+ Ok(documents.len())
+ }
+
+ async fn wait_for_task(&self, client: &Client, task: TaskInfo) ->
Result<(), Error> {
+ if !self.config.wait_for_tasks {
+ return Ok(());
+ }
+
+ self.wait_for_task_completion(client, task).await
+ }
+
+ async fn wait_for_index_creation_task(
+ &self,
+ client: &Client,
+ task: TaskInfo,
+ ) -> Result<(), Error> {
+ let task = self
+ .wait_for_task_status(client, task, self.config.max_open_retries)
+ .await?;
+
+ if task.is_success() {
+ return Ok(());
+ }
+
+ if task.is_failure() {
+ let failure = task.unwrap_failure();
+ if failure.error_code == MeilisearchErrorCode::IndexAlreadyExists {
+ return Ok(());
+ }
+ return Err(Error::PermanentHttpError(format!(
+ "Meilisearch task failed: {}",
+ failure
+ )));
+ }
+
+ Err(Error::HttpRequestFailed(
+ "Meilisearch task did not reach a terminal state".to_string(),
+ ))
+ }
+
+ async fn wait_for_task_completion(&self, client: &Client, task: TaskInfo)
-> Result<(), Error> {
+ let task = self
+ .wait_for_task_status(client, task, self.config.max_retries)
+ .await?;
+
+ if task.is_success() {
+ return Ok(());
+ }
+
+ if task.is_failure() {
+ let failure = task.unwrap_failure();
+ return Err(Error::PermanentHttpError(format!(
+ "Meilisearch task failed: {}",
+ failure
+ )));
+ }
+
+ Err(Error::HttpRequestFailed(
+ "Meilisearch task did not reach a terminal state".to_string(),
+ ))
+ }
+
+ async fn wait_for_task_status(
+ &self,
+ client: &Client,
+ task: TaskInfo,
+ max_get_task_retries: u32,
+ ) -> Result<Task, Error> {
+ let task_uid = task.get_task_uid();
+ let started = Instant::now();
+
+ loop {
+ let remaining =
self.config.task_timeout.saturating_sub(started.elapsed());
+ if remaining.is_zero() {
+ break;
+ }
+
+ let status = match tokio::time::timeout(
+ remaining,
+ self.retry_sdk_operation_with_retries(
+ "get task status",
+ max_get_task_retries,
+ || client.get_task(TaskUid(task_uid)),
+ ),
+ )
+ .await
+ {
+ Ok(status) => status?,
+ Err(_) => break,
+ };
+
+ if status.is_success() || status.is_failure() {
+ return Ok(status);
+ }
+
+ let remaining =
self.config.task_timeout.saturating_sub(started.elapsed());
+ if remaining.is_zero() {
+ break;
+ }
+ sleep(cmp::min(self.config.task_poll_interval, remaining)).await;
+ }
+
+ Err(Error::HttpRequestFailed(format!(
+ "Meilisearch task {task_uid} timed out after {:?}",
+ self.config.task_timeout
+ )))
+ }
+
+ async fn retry_sdk_operation<T, Fut, Op>(
+ &self,
+ operation: &str,
+ operation_fn: Op,
+ ) -> Result<T, Error>
+ where
+ Op: FnMut() -> Fut,
+ Fut: Future<Output = Result<T, MeilisearchSdkError>>,
+ {
+ self.retry_sdk_operation_with_retries(operation,
self.config.max_retries, operation_fn)
+ .await
+ }
+
+ async fn retry_sdk_open_operation<T, Fut, Op>(
+ &self,
+ operation: &str,
+ operation_fn: Op,
+ ) -> Result<T, Error>
+ where
+ Op: FnMut() -> Fut,
+ Fut: Future<Output = Result<T, MeilisearchSdkError>>,
+ {
+ self.retry_sdk_operation_with_retries(operation,
self.config.max_open_retries, operation_fn)
+ .await
+ }
+
+ async fn retry_sdk_operation_with_retries<T, Fut, Op>(
+ &self,
+ operation: &str,
+ max_retries: u32,
+ mut operation_fn: Op,
+ ) -> Result<T, Error>
+ where
+ Op: FnMut() -> Fut,
+ Fut: Future<Output = Result<T, MeilisearchSdkError>>,
+ {
+ let started = Instant::now();
+ let mut retries = 0u32;
+
+ loop {
+ let remaining =
self.config.timeout.saturating_sub(started.elapsed());
+ if remaining.is_zero() {
+ return Err(Error::HttpRequestFailed(format!(
+ "Meilisearch {operation} timed out after {:?}",
+ self.config.timeout
+ )));
+ }
+
+ let result = tokio::time::timeout(remaining, operation_fn()).await;
+ match result {
+ Ok(Ok(value)) => return Ok(value),
+ Ok(Err(error)) => {
+ let should_retry = retries < max_retries &&
is_transient_sdk_error(&error);
+ if !should_retry {
+ return Err(map_sdk_error(error));
+ }
+ retries += 1;
+ let delay = jitter(exponential_backoff(
+ self.config.retry_delay,
+ retries,
+ self.config.max_retry_delay,
+ ));
+ warn!(
+ "Meilisearch {operation} failed (retry
{retries}/{max_retries}): {error}. Retrying in {delay:?}..."
+ );
+ let remaining =
self.config.timeout.saturating_sub(started.elapsed());
+ if remaining.is_zero() {
+ return Err(map_sdk_error(error));
+ }
+ sleep(cmp::min(delay, remaining)).await;
+ }
+ Err(_) => {
+ if retries >= max_retries {
+ return Err(Error::HttpRequestFailed(format!(
+ "Meilisearch {operation} timed out after {:?}",
+ self.config.timeout
+ )));
+ }
+ retries += 1;
+ let delay = jitter(exponential_backoff(
+ self.config.retry_delay,
+ retries,
+ self.config.max_retry_delay,
+ ));
+ warn!(
+ "Meilisearch {operation} timed out after {:?} (retry
{retries}/{max_retries}). Retrying in {delay:?}...",
+ self.config.timeout
+ );
+ let remaining =
self.config.timeout.saturating_sub(started.elapsed());
+ if remaining.is_zero() {
+ return Err(Error::HttpRequestFailed(format!(
+ "Meilisearch {operation} timed out after {:?}",
+ self.config.timeout
+ )));
+ }
+ sleep(cmp::min(delay, remaining)).await;
+ }
+ }
+ }
+ }
+
+ async fn record_errors(&self, errors: usize) {
+ let mut state = self.state.lock().await;
+ state.errors_count += errors;
+ }
+}
+
+#[async_trait]
+impl Sink for MeilisearchSink {
+ async fn open(&mut self) -> Result<(), Error> {
+ self.validate_config()?;
+ info!(
+ "Opening Meilisearch sink connector with ID: {} for URL: {},
index: {}",
+ self.id,
+ sanitize_url_for_log(&self.config.url),
+ self.config.index
+ );
+ if self.config.document_action == MeilisearchDocumentAction::Update {
+ warn!(
+ "Meilisearch sink connector with ID: {} is using
document_action=update. Internal retries and ambiguous task outcomes can apply
non-idempotent updates more than once.",
+ self.id
+ );
+ }
+ if !self.config.wait_for_tasks {
+ warn!(
+ "Meilisearch sink connector with ID: {} is opening with
wait_for_tasks=false. Submitted document tasks may still be in flight or fail
after offsets are committed; this mode does not provide durability.",
+ self.id
+ );
+ }
+
+ let client = self.create_client()?;
+ self.check_connectivity(&client).await?;
+ self.ensure_index_exists(&client).await?;
+
+ self.client = Some(client);
+ info!(
+ "Successfully opened Meilisearch sink connector with ID: {}",
+ self.id
+ );
+ Ok(())
+ }
+
+ async fn consume(
+ &self,
+ topic_metadata: &TopicMetadata,
+ messages_metadata: MessagesMetadata,
+ messages: Vec<ConsumedMessage>,
+ ) -> Result<(), Error> {
+ let mut state = self.state.lock().await;
+ state.invocations_count += 1;
+ let invocation = state.invocations_count;
+ drop(state);
+
+ info!(
+ "Meilisearch sink with ID: {} received: {} messages, schema: {},
stream: {}, topic: {}, partition: {}, offset: {}, invocation: {}",
+ self.id,
+ messages.len(),
+ messages_metadata.schema,
+ topic_metadata.stream,
+ topic_metadata.topic,
+ messages_metadata.partition_id,
+ messages_metadata.current_offset,
+ invocation
+ );
+
+ let client = self
+ .client
+ .as_ref()
+ .ok_or_else(|| Error::Connection("Meilisearch client not
initialized".to_string()))?;
+
+ let messages_count = messages.len();
+ let mut documents = Vec::with_capacity(messages.len());
+ let mut invalid_records = 0usize;
+ for message in messages {
+ match self.prepare_document(topic_metadata, &messages_metadata,
message) {
+ Ok(document) => documents.push(document),
+ Err(Error::InvalidRecordValue(reason)) => {
+ invalid_records += 1;
+ warn!(
+ "Dropping invalid Meilisearch sink record for
connector ID: {}, reason: {}",
+ self.id, reason
+ );
+ }
+ Err(error) => return Err(error),
+ }
+ }
+ if invalid_records > 0 {
+ self.record_errors(invalid_records).await;
+ }
+
+ if documents.is_empty() {
+ return Ok(());
+ }
+
+ match self.index_documents(client, documents).await {
+ Ok(accepted) => {
+ let mut state = self.state.lock().await;
+ state.documents_enqueued += accepted;
+ if self.config.wait_for_tasks {
+ state.documents_confirmed += accepted;
+ }
+ info!(
+ "Accepted {} of {} messages into Meilisearch index '{}'",
+ accepted, messages_count, self.config.index
+ );
+ Ok(())
+ }
+ Err(partial_error) => {
+ let mut state = self.state.lock().await;
+ state.documents_enqueued += partial_error.accepted;
+ if self.config.wait_for_tasks {
+ state.documents_confirmed += partial_error.accepted;
+ }
+ state.errors_count += partial_error.failed;
+ drop(state);
+ error!(
+ "Failed to index Meilisearch sink batch for connector ID:
{}, index: {}, accepted: {}, failed: {}, error: {}",
+ self.id,
+ self.config.index,
+ partial_error.accepted,
+ partial_error.failed,
+ partial_error.error
+ );
+ Err(partial_error.error)
+ }
+ }
+ }
+
+ async fn close(&mut self) -> Result<(), Error> {
+ let state = self.state.lock().await;
+ if self.config.wait_for_tasks {
+ info!(
+ "Meilisearch sink connector with ID: {} is closing. Stats: {}
invocations, {} documents enqueued, {} documents confirmed, {} errors",
+ self.id,
+ state.invocations_count,
+ state.documents_enqueued,
+ state.documents_confirmed,
+ state.errors_count
+ );
+ } else {
+ warn!(
+ "Meilisearch sink connector with ID: {} is closing with
wait_for_tasks=false. Submitted document tasks may still be in flight or fail
after offsets are committed.",
+ self.id
+ );
+ info!(
+ "Meilisearch sink connector with ID: {} is closing. Stats: {}
invocations, {} documents enqueued, documents confirmed unavailable
(wait_for_tasks=false), {} errors",
+ self.id, state.invocations_count, state.documents_enqueued,
state.errors_count
+ );
+ }
+ drop(state);
+
+ self.client = None;
+ info!("Meilisearch sink connector with ID: {} is closed.", self.id);
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+fn generated_document_id(
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ message: &ConsumedMessage,
+) -> String {
+ generated_document_id_from_parts(
+ topic_metadata,
+ messages_metadata,
+ message.offset,
+ message.id,
+ )
+ .expect("test generated ID components should serialize")
+}
+
+fn generated_document_id_from_parts(
+ topic_metadata: &TopicMetadata,
+ messages_metadata: &MessagesMetadata,
+ offset: u64,
+ id: u128,
+) -> Result<String, Error> {
+ let components = json!([
+ topic_metadata.stream.as_str(),
+ topic_metadata.topic.as_str(),
+ messages_metadata.partition_id,
+ offset,
+ id.to_string()
+ ]);
+ let encoded = serde_json::to_vec(&components)
+ .map(|bytes| general_purpose::URL_SAFE_NO_PAD.encode(bytes))
+ .map_err(|error| {
+ Error::Serialization(format!(
+ "Failed to serialize generated document ID: {error}"
+ ))
+ })?;
+ Ok(format!("iggy_{encoded}"))
+}
+
+fn upsert_metadata_field(object: &mut Map<String, Value>, field: &str, value:
Value) {
+ if object.insert(field.to_string(), value).is_some() {
+ debug!(
+ "Document already contains Meilisearch metadata field '{field}',
overwriting with connector provenance"
+ );
+ }
+}
+
+fn owned_value_into_serde_json(value: simd_json::OwnedValue) -> Value {
+ match value {
+ simd_json::OwnedValue::Static(node) => match node {
+ simd_json::StaticNode::Null => Value::Null,
+ simd_json::StaticNode::Bool(value) => Value::Bool(value),
+ simd_json::StaticNode::I64(value) => Value::Number(value.into()),
+ simd_json::StaticNode::U64(value) => Value::Number(value.into()),
+ simd_json::StaticNode::F64(value) =>
serde_json::Number::from_f64(value)
+ .map(Value::Number)
+ .unwrap_or(Value::Null),
+ },
+ simd_json::OwnedValue::String(value) => Value::String(value),
+ simd_json::OwnedValue::Array(values) => Value::Array(
+ values
+ .into_iter()
+ .map(owned_value_into_serde_json)
+ .collect(),
+ ),
+ simd_json::OwnedValue::Object(values) => Value::Object(
+ values
+ .into_iter()
+ .map(|(key, value)| (key, owned_value_into_serde_json(value)))
+ .collect(),
+ ),
+ }
+}
+
+fn sanitize_url_for_log(raw: &str) -> String {
+ let normalized = normalize_host(raw).unwrap_or_else(|_|
raw.trim().to_string());
+ let Ok(mut url) = Url::parse(&normalized) else {
+ return "<invalid-url>".to_string();
+ };
+
+ if !url.username().is_empty() {
+ let _ = url.set_username("");
+ }
+ if url.password().is_some() {
+ let _ = url.set_password(None);
+ }
+ url.to_string().trim_end_matches('/').to_string()
+}
+
+fn normalize_host(raw: &str) -> Result<String, Error> {
+ let trimmed = raw.trim();
+ if trimmed.is_empty() {
+ return Err(Error::Connection(
+ "Invalid Meilisearch URL: host cannot be empty".to_string(),
+ ));
+ }
+
+ let with_scheme = if trimmed.starts_with("http://";) ||
trimmed.starts_with("https://";) {
+ trimmed.to_string()
+ } else {
+ format!("http://{trimmed}";)
+ };
+ let url = Url::parse(&with_scheme)
+ .map_err(|error| Error::Connection(format!("Invalid Meilisearch URL:
{error}")))?;
+ if url.path() != "/" || url.query().is_some() || url.fragment().is_some() {
+ warn!("Ignoring path, query, or fragment from Meilisearch URL");
+ }
+ let mut base_url = url;
+ base_url.set_path("");
+ base_url.set_query(None);
+ base_url.set_fragment(None);
+ Ok(base_url.as_str().trim_end_matches('/').to_string())
+}
+
+fn warn_if_api_key_uses_insecure_http(raw: &str, normalized: &str,
has_api_key: bool) {
+ if !has_api_key {
+ return;
+ }
+
+ let Ok(url) = Url::parse(normalized) else {
+ return;
+ };
+ if url.scheme() != "http" {
+ return;
+ }
+ let Some(host) = url.host_str() else {
+ return;
+ };
+ if is_loopback_host(host) {
+ return;
+ }
+
+ let scheme_hint = if raw.trim().starts_with("http://";) {
+ "explicit http://";
+ } else {
+ "implicit http://";
+ };
+ warn!(
+ "Meilisearch API key is configured with {scheme_hint} for non-loopback
host '{host}'. Credentials will be sent without TLS; use https:// unless this
is intentional."
+ );
+}
+
+fn is_loopback_host(host: &str) -> bool {
+ host.eq_ignore_ascii_case("localhost")
+ || host
+ .parse::<IpAddr>()
+ .map(|address| address.is_loopback())
+ .unwrap_or(false)
+}
+
+#[derive(Debug)]
+struct PartialIndexError {
+ accepted: usize,
+ failed: usize,
+ error: Error,
+}
+
+#[derive(Clone, Copy)]
+struct TaskUid(u32);
+
+impl AsRef<u32> for TaskUid {
+ fn as_ref(&self) -> &u32 {
+ &self.0
+ }
+}
+
+fn is_index_not_found(error: &MeilisearchSdkError) -> bool {
+ matches!(
+ error,
+ MeilisearchSdkError::Meilisearch(meilisearch_error)
+ if meilisearch_error.error_code ==
MeilisearchErrorCode::IndexNotFound
+ )
+}
+
+fn is_transient_sdk_error(error: &MeilisearchSdkError) -> bool {
+ match error {
+ MeilisearchSdkError::Meilisearch(meilisearch_error) => {
+ meilisearch_error.error_type == MeilisearchErrorType::Internal
+ }
+ MeilisearchSdkError::MeilisearchCommunication(communication_error) => {
+ communication_error.status_code == 0
+ || communication_error.status_code == 429
+ || communication_error.status_code >= 500
+ }
+ MeilisearchSdkError::HttpError(_) | MeilisearchSdkError::Timeout =>
true,
+ _ => false,
+ }
+}
+
+fn map_sdk_error(error: MeilisearchSdkError) -> Error {
+ match error {
+ MeilisearchSdkError::Meilisearch(meilisearch_error) => {
+ if meilisearch_error.error_type == MeilisearchErrorType::Internal {
+ Error::HttpRequestFailed(meilisearch_error.to_string())
+ } else {
+ Error::PermanentHttpError(meilisearch_error.to_string())
+ }
+ }
+ MeilisearchSdkError::MeilisearchCommunication(communication_error) => {
+ if communication_error.status_code == 0
+ || communication_error.status_code == 429
+ || communication_error.status_code >= 500
+ {
+ Error::HttpRequestFailed(communication_error.to_string())
+ } else {
+ Error::PermanentHttpError(communication_error.to_string())
+ }
+ }
+ MeilisearchSdkError::ParseError(error) => {
+ Error::Serialization(format!("Invalid Meilisearch response:
{error}"))
+ }
+ MeilisearchSdkError::Timeout => {
+ Error::HttpRequestFailed("Meilisearch task timed out".to_string())
+ }
+ MeilisearchSdkError::HttpError(error) =>
Error::HttpRequestFailed(error.to_string()),
+ other => Error::HttpRequestFailed(other.to_string()),
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use iggy_connector_sdk::Schema;
+
+ fn topic_metadata() -> TopicMetadata {
+ TopicMetadata {
+ stream: "orders.stream".to_string(),
+ topic: "created/topic".to_string(),
+ }
+ }
+
+ fn messages_metadata() -> MessagesMetadata {
+ MessagesMetadata {
+ partition_id: 7,
+ current_offset: 10,
+ schema: Schema::Json,
+ }
+ }
+
+ fn message(payload: Payload) -> ConsumedMessage {
+ ConsumedMessage {
+ id: 42,
+ offset: 11,
+ checksum: 12,
+ timestamp: 13,
+ origin_timestamp: 14,
+ headers: None,
+ payload,
+ }
+ }
+
+ fn sink_with_config(config: MeilisearchSinkConfig) -> MeilisearchSink {
+ MeilisearchSink::new(1, config)
+ }
+
+ fn base_config() -> MeilisearchSinkConfig {
+ MeilisearchSinkConfig {
+ url: "http://localhost:7700".to_string(),
+ index: "messages".to_string(),
+ api_key: None,
+ primary_key: None,
+ document_action: None,
+ create_index_if_not_exists: None,
+ include_metadata: None,
+ batch_size: None,
+ timeout: None,
+ wait_for_tasks: None,
+ task_timeout: None,
+ task_poll_interval: None,
+ max_retries: None,
+ retry_delay: None,
+ max_retry_delay: None,
+ max_open_retries: None,
+ }
+ }
+
+ #[test]
+ fn swaps_retry_delays_when_retry_delay_exceeds_max_retry_delay() {
+ let mut config = base_config();
+ config.retry_delay = Some("10s".to_string());
+ config.max_retry_delay = Some("1s".to_string());
+
+ let sink = sink_with_config(config);
+
+ assert_eq!(sink.config.retry_delay, Duration::from_secs(1));
+ assert_eq!(sink.config.max_retry_delay, Duration::from_secs(10));
+ }
+
+ #[test]
+ fn generated_ids_use_meilisearch_safe_characters() {
+ let id = generated_document_id(
+ &topic_metadata(),
+ &messages_metadata(),
+ &message(Payload::Text("x".to_string())),
+ );
+
+ assert!(id.starts_with("iggy_"));
+ assert!(
+ id.chars()
+ .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
+ );
+ }
+
+ #[test]
+ fn generated_ids_do_not_collapse_sanitized_names() {
+ let first_topic = TopicMetadata {
+ stream: "orders.stream".to_string(),
+ topic: "created/topic".to_string(),
+ };
+ let second_topic = TopicMetadata {
+ stream: "orders/stream".to_string(),
+ topic: "created.topic".to_string(),
+ };
+
+ let first = generated_document_id(
+ &first_topic,
+ &messages_metadata(),
+ &message(Payload::Text("x".to_string())),
+ );
+ let second = generated_document_id(
+ &second_topic,
+ &messages_metadata(),
+ &message(Payload::Text("x".to_string())),
+ );
+
+ assert_ne!(first, second);
+ }
+
+ #[test]
+ fn generated_ids_support_u128_message_ids() {
+ let mut message = message(Payload::Text("x".to_string()));
+ message.id = u128::MAX;
+
+ let id = generated_document_id(&topic_metadata(),
&messages_metadata(), &message);
+
+ assert!(id.starts_with("iggy_"));
+ }
+
+ #[test]
+ fn injects_default_primary_key_and_metadata() {
+ let sink = sink_with_config(base_config());
+ let payload = Payload::Json(simd_json::json!({
+ "name": "Alice"
+ }));
+ let message = message(payload);
+ let expected_id = generated_document_id(&topic_metadata(),
&messages_metadata(), &message);
+
+ let document = sink
+ .prepare_document(&topic_metadata(), &messages_metadata(), message)
+ .expect("prepare document");
+
+ assert_eq!(document["name"], "Alice");
+ assert_eq!(document["iggy_id"], expected_id);
+ assert_eq!(document["iggy_offset"], 11);
+ assert_eq!(document["iggy_stream"], "orders.stream");
+ assert_eq!(document["iggy_topic"], "created/topic");
+ assert_eq!(document["iggy_checksum"], "12");
+ }
+
+ #[test]
+ fn preserves_existing_configured_primary_key() {
+ let mut config = base_config();
+ config.primary_key = Some("id".to_string());
+ let sink = sink_with_config(config);
+ let payload = Payload::Json(simd_json::json!({
+ "id": "existing",
+ "name": "Alice"
+ }));
+ let message = message(payload);
+ let expected_id = generated_document_id(&topic_metadata(),
&messages_metadata(), &message);
+
+ let document = sink
+ .prepare_document(&topic_metadata(), &messages_metadata(), message)
+ .expect("prepare document");
+
+ assert_eq!(document["id"], "existing");
+ assert_eq!(document["iggy_id"], expected_id);
+ }
+
+ #[test]
+ fn overwrites_reserved_metadata_fields() {
+ let sink = sink_with_config(base_config());
+ let payload = Payload::Json(simd_json::json!({
+ "name": "Alice",
+ "iggy_offset": 999,
+ "iggy_stream": "user-stream",
+ "iggy_checksum": 999
+ }));
+
+ let document = sink
+ .prepare_document(&topic_metadata(), &messages_metadata(),
message(payload))
+ .expect("prepare document");
+
+ assert_eq!(document["iggy_offset"], 11);
+ assert_eq!(document["iggy_stream"], "orders.stream");
+ assert_eq!(document["iggy_checksum"], "12");
+ }
+
+ #[test]
+ fn omits_metadata_when_include_metadata_is_false() {
+ let mut config = base_config();
+ config.include_metadata = Some(false);
+ let sink = sink_with_config(config);
+ let payload = Payload::Json(simd_json::json!({
+ "name": "Alice"
+ }));
+
+ let document = sink
+ .prepare_document(&topic_metadata(), &messages_metadata(),
message(payload))
+ .expect("prepare document");
+
+ assert_eq!(document["name"], "Alice");
+ assert!(document["iggy_id"].as_str().is_some());
+ assert!(document.get("iggy_offset").is_none());
+ assert!(document.get("iggy_stream").is_none());
+ }
+
+ #[test]
+ fn wraps_non_object_json_payloads() {
+ let sink = sink_with_config(base_config());
+ let message = message(Payload::Json(simd_json::json!(["a", "b"])));
+ let expected_id = generated_document_id(&topic_metadata(),
&messages_metadata(), &message);
+
+ let document = sink
+ .prepare_document(&topic_metadata(), &messages_metadata(), message)
+ .expect("prepare document");
+
+ assert_eq!(document["value"], json!(["a", "b"]));
+ assert_eq!(document["iggy_id"], expected_id);
+ }
+
+ #[test]
+ fn raw_payloads_are_base64_encoded_when_not_json() {
+ let sink = sink_with_config(base_config());
+
+ let document = sink
+ .prepare_document(
+ &topic_metadata(),
+ &messages_metadata(),
+ message(Payload::Raw(vec![0, 1, 2, 3])),
+ )
+ .expect("prepare document");
+
+ assert_eq!(document["data"], "AAECAw==");
+ assert_eq!(document["data_encoding"], ENCODING_BASE64);
+ }
+
+ #[test]
+ fn raw_payloads_preserve_original_bytes_when_json_parse_mutates_buffer() {
+ let sink = sink_with_config(base_config());
+ let bytes = b"[1,2,3".to_vec();
+ let expected = general_purpose::STANDARD.encode(&bytes);
+
+ let document = sink
+ .prepare_document(
+ &topic_metadata(),
+ &messages_metadata(),
+ message(Payload::Raw(bytes)),
+ )
+ .expect("prepare document");
+
+ assert_eq!(document["data"], expected);
+ assert_eq!(document["data_encoding"], ENCODING_BASE64);
+ }
+
+ #[test]
+ fn sanitize_url_should_redact_credentials_without_scheme() {
+ let url = sanitize_url_for_log("user:pass@localhost:7700/indexes");
+ assert_eq!(url, "http://localhost:7700";);
+ }
+
+ #[test]
+ fn normalize_host_should_strip_path_query_and_fragment() {
+ let url =
+
normalize_host("https://localhost:7700/path?foo=bar#section";).expect("normalize
host");
+
+ assert_eq!(url, "https://localhost:7700";);
+ }
+
+ #[test]
+ fn validate_config_should_reject_empty_index() {
+ let mut config = base_config();
+ config.index = " ".to_string();
+ let sink = sink_with_config(config);
+
+ let error = sink.validate_config().expect_err("empty index should
fail");
+
+ assert!(matches!(error, Error::InvalidConfigValue(_)));
+ }
+
+ #[test]
+ fn loopback_host_detection_allows_local_addresses() {
+ assert!(is_loopback_host("localhost"));
+ assert!(is_loopback_host("127.0.0.1"));
+ assert!(is_loopback_host("::1"));
+ assert!(!is_loopback_host("meili.prod"));
+ }
+
+ #[test]
+ fn unsupported_payloads_return_error() {
+ let sink = sink_with_config(base_config());
+ let error = sink
+ .prepare_document(
+ &topic_metadata(),
+ &messages_metadata(),
+ message(Payload::Avro(vec![1, 2, 3])),
+ )
+ .expect_err("unsupported payload should fail");
+
+ assert!(matches!(error, Error::InvalidRecordValue(_)));
+ }
+}
diff --git
a/core/integration/tests/connectors/fixtures/meilisearch/container.rs
b/core/integration/tests/connectors/fixtures/meilisearch/container.rs
new file mode 100644
index 000000000..55cadbe21
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/meilisearch/container.rs
@@ -0,0 +1,180 @@
+// 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.
+
+use crate::connectors::fixtures;
+use integration::harness::TestBinaryError;
+use reqwest_middleware::ClientWithMiddleware as HttpClient;
+use reqwest_retry::RetryTransientMiddleware;
+use reqwest_retry::policies::ExponentialBackoff;
+use serde::Deserialize;
+use std::time::Duration;
+use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy;
+use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor};
+use testcontainers_modules::testcontainers::runners::AsyncRunner;
+use testcontainers_modules::testcontainers::{ContainerAsync, GenericImage,
ImageExt};
+use tokio::time::sleep;
+use tracing::info;
+use uuid::Uuid;
+
+const MEILISEARCH_IMAGE: &str = "getmeili/meilisearch";
+const MEILISEARCH_TAG: &str = "v1.13";
+const MEILISEARCH_PORT: u16 = 7700;
+const MEILISEARCH_HEALTH_ENDPOINT: &str = "/health";
+pub const TEST_INDEX: &str = "iggy_messages";
+const POLL_ATTEMPTS: usize = 100;
+const POLL_INTERVAL_MS: u64 = 50;
+
+#[derive(Debug, Deserialize)]
+pub struct MeilisearchDocumentsResponse {
+ pub results: Vec<serde_json::Value>,
+}
+
+pub struct MeilisearchContainer {
+ #[allow(dead_code)]
+ container: ContainerAsync<GenericImage>,
+ pub base_url: String,
+}
+
+impl MeilisearchContainer {
+ pub async fn start() -> Result<Self, TestBinaryError> {
+ let unique_network = format!("iggy-meilisearch-{}", Uuid::new_v4());
+
+ let container = GenericImage::new(MEILISEARCH_IMAGE, MEILISEARCH_TAG)
+ .with_exposed_port(MEILISEARCH_PORT.tcp())
+ .with_wait_for(WaitFor::http(
+ HttpWaitStrategy::new(MEILISEARCH_HEALTH_ENDPOINT)
+ .with_port(MEILISEARCH_PORT.tcp())
+ .with_expected_status_code(200u16),
+ ))
+ .with_network(unique_network)
+
.with_container_name(fixtures::unique_container_name("meilisearch"))
+ .with_env_var("MEILI_ENV", "development")
+ .with_mapped_port(0, MEILISEARCH_PORT.tcp())
+ .start()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "MeilisearchContainer".to_string(),
+ message: format!("Failed to start container: {e}"),
+ })?;
+
+ info!("Started Meilisearch container");
+
+ let mapped_port = container
+ .ports()
+ .await
+ .map_err(|e| TestBinaryError::FixtureSetup {
+ fixture_type: "MeilisearchContainer".to_string(),
+ message: format!("Failed to get ports: {e}"),
+ })?
+ .map_to_host_port_ipv4(MEILISEARCH_PORT)
+ .ok_or_else(|| TestBinaryError::FixtureSetup {
+ fixture_type: "MeilisearchContainer".to_string(),
+ message: "No mapping for Meilisearch port".to_string(),
+ })?;
+
+ let base_url = format!("http://localhost:{mapped_port}";);
+ info!("Meilisearch container available at {base_url}");
+
+ Ok(Self {
+ container,
+ base_url,
+ })
+ }
+}
+
+pub fn create_http_client() -> HttpClient {
+ let retry_policy = ExponentialBackoff::builder().build_with_max_retries(3);
+ let client = reqwest::Client::builder()
+ .timeout(Duration::from_secs(30))
+ .build()
+ .expect("Failed to build HTTP client");
+ reqwest_middleware::ClientBuilder::new(client)
+ .with(RetryTransientMiddleware::new_with_policy(retry_policy))
+ .build()
+}
+
+pub trait MeilisearchOps: Sync {
+ fn container(&self) -> &MeilisearchContainer;
+ fn http_client(&self) -> &HttpClient;
+
+ fn list_documents(
+ &self,
+ index_name: &str,
+ ) -> impl std::future::Future<Output = Result<Vec<serde_json::Value>,
TestBinaryError>> + Send
+ {
+ async move {
+ let url = format!(
+ "{}/indexes/{}/documents",
+ self.container().base_url,
+ index_name
+ );
+ let response = self
+ .http_client()
+ .get(&url)
+ .query(&[("limit", "100")])
+ .send()
+ .await
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to list Meilisearch documents:
{e}"),
+ })?;
+
+ if !response.status().is_success() {
+ let status = response.status();
+ let body = response.text().await.unwrap_or_default();
+ return Err(TestBinaryError::InvalidState {
+ message: format!(
+ "Failed to list Meilisearch documents:
status={status}, body={body}"
+ ),
+ });
+ }
+
+ response
+ .json::<MeilisearchDocumentsResponse>()
+ .await
+ .map(|documents| documents.results)
+ .map_err(|e| TestBinaryError::InvalidState {
+ message: format!("Failed to parse Meilisearch documents
response: {e}"),
+ })
+ }
+ }
+
+ fn wait_for_documents(
+ &self,
+ index_name: &str,
+ expected_count: usize,
+ ) -> impl std::future::Future<Output = Result<Vec<serde_json::Value>,
TestBinaryError>> + Send
+ {
+ async move {
+ let mut last_count = 0usize;
+ for _ in 0..POLL_ATTEMPTS {
+ if let Ok(documents) = self.list_documents(index_name).await {
+ last_count = documents.len();
+ if documents.len() >= expected_count {
+ return Ok(documents);
+ }
+ }
+ sleep(Duration::from_millis(POLL_INTERVAL_MS)).await;
+ }
+
+ Err(TestBinaryError::InvalidState {
+ message: format!(
+ "Expected {expected_count} Meilisearch documents, found
{last_count} after {POLL_ATTEMPTS} attempts"
+ ),
+ })
+ }
+ }
+}
diff --git a/core/integration/tests/connectors/fixtures/meilisearch/mod.rs
b/core/integration/tests/connectors/fixtures/meilisearch/mod.rs
new file mode 100644
index 000000000..09e0b9d91
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/meilisearch/mod.rs
@@ -0,0 +1,22 @@
+// 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.
+
+mod container;
+mod sink;
+
+pub use container::{MeilisearchOps, TEST_INDEX};
+pub use sink::MeilisearchSinkFixture;
diff --git a/core/integration/tests/connectors/fixtures/meilisearch/sink.rs
b/core/integration/tests/connectors/fixtures/meilisearch/sink.rs
new file mode 100644
index 000000000..23f4ce7be
--- /dev/null
+++ b/core/integration/tests/connectors/fixtures/meilisearch/sink.rs
@@ -0,0 +1,90 @@
+// 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.
+
+use super::container::{MeilisearchContainer, MeilisearchOps, TEST_INDEX,
create_http_client};
+use async_trait::async_trait;
+use integration::harness::{TestBinaryError, TestFixture, seeds};
+use reqwest_middleware::ClientWithMiddleware as HttpClient;
+use std::collections::HashMap;
+
+const ENV_SINK_URL: &str =
"IGGY_CONNECTORS_SINK_MEILISEARCH_PLUGIN_CONFIG_URL";
+const ENV_SINK_INDEX: &str =
"IGGY_CONNECTORS_SINK_MEILISEARCH_PLUGIN_CONFIG_INDEX";
+const ENV_SINK_PRIMARY_KEY: &str =
"IGGY_CONNECTORS_SINK_MEILISEARCH_PLUGIN_CONFIG_PRIMARY_KEY";
+const ENV_SINK_TASK_POLL_INTERVAL: &str =
+ "IGGY_CONNECTORS_SINK_MEILISEARCH_PLUGIN_CONFIG_TASK_POLL_INTERVAL";
+const ENV_SINK_TASK_TIMEOUT: &str =
"IGGY_CONNECTORS_SINK_MEILISEARCH_PLUGIN_CONFIG_TASK_TIMEOUT";
+const ENV_SINK_STREAMS_0_STREAM: &str =
"IGGY_CONNECTORS_SINK_MEILISEARCH_STREAMS_0_STREAM";
+const ENV_SINK_STREAMS_0_TOPICS: &str =
"IGGY_CONNECTORS_SINK_MEILISEARCH_STREAMS_0_TOPICS";
+const ENV_SINK_STREAMS_0_SCHEMA: &str =
"IGGY_CONNECTORS_SINK_MEILISEARCH_STREAMS_0_SCHEMA";
+const ENV_SINK_STREAMS_0_CONSUMER_GROUP: &str =
+ "IGGY_CONNECTORS_SINK_MEILISEARCH_STREAMS_0_CONSUMER_GROUP";
+const ENV_SINK_PATH: &str = "IGGY_CONNECTORS_SINK_MEILISEARCH_PATH";
+
+pub struct MeilisearchSinkFixture {
+ container: MeilisearchContainer,
+ http_client: HttpClient,
+}
+
+impl MeilisearchOps for MeilisearchSinkFixture {
+ fn container(&self) -> &MeilisearchContainer {
+ &self.container
+ }
+
+ fn http_client(&self) -> &HttpClient {
+ &self.http_client
+ }
+}
+
+#[async_trait]
+impl TestFixture for MeilisearchSinkFixture {
+ async fn setup() -> Result<Self, TestBinaryError> {
+ let container = MeilisearchContainer::start().await?;
+ let http_client = create_http_client();
+
+ Ok(Self {
+ container,
+ http_client,
+ })
+ }
+
+ fn connectors_runtime_envs(&self) -> HashMap<String, String> {
+ HashMap::from([
+ (ENV_SINK_URL.to_string(), self.container.base_url.clone()),
+ (ENV_SINK_INDEX.to_string(), TEST_INDEX.to_string()),
+ (ENV_SINK_PRIMARY_KEY.to_string(), "iggy_id".to_string()),
+ (ENV_SINK_TASK_TIMEOUT.to_string(), "10s".to_string()),
+ (ENV_SINK_TASK_POLL_INTERVAL.to_string(), "25ms".to_string()),
+ (
+ ENV_SINK_STREAMS_0_STREAM.to_string(),
+ seeds::names::STREAM.to_string(),
+ ),
+ (
+ ENV_SINK_STREAMS_0_TOPICS.to_string(),
+ format!("[{}]", seeds::names::TOPIC),
+ ),
+ (ENV_SINK_STREAMS_0_SCHEMA.to_string(), "json".to_string()),
+ (
+ ENV_SINK_STREAMS_0_CONSUMER_GROUP.to_string(),
+ "meilisearch_sink".to_string(),
+ ),
+ (
+ ENV_SINK_PATH.to_string(),
+
"../../target/debug/libiggy_connector_meilisearch_sink".to_string(),
+ ),
+ ])
+ }
+}
diff --git a/core/integration/tests/connectors/fixtures/mod.rs
b/core/integration/tests/connectors/fixtures/mod.rs
index 6d75fbf97..e4992d678 100644
--- a/core/integration/tests/connectors/fixtures/mod.rs
+++ b/core/integration/tests/connectors/fixtures/mod.rs
@@ -24,6 +24,7 @@ mod elasticsearch;
mod http;
mod iceberg;
mod influxdb;
+mod meilisearch;
mod mongodb;
mod postgres;
mod quickwit;
@@ -68,6 +69,7 @@ pub use influxdb::{
InfluxDbSinkNoMetadataFixture, InfluxDbSinkNsPrecisionFixture,
InfluxDbSinkTextFixture,
InfluxDbSourceFixture, InfluxDbSourceRawFixture, InfluxDbSourceTextFixture,
};
+pub use meilisearch::{MeilisearchOps, MeilisearchSinkFixture, TEST_INDEX};
pub use mongodb::{
MongoDbOps, MongoDbSinkAutoCreateFixture, MongoDbSinkBatchFixture,
MongoDbSinkFailpointFixture,
MongoDbSinkFixture, MongoDbSinkJsonFixture, MongoDbSinkWriteConcernFixture,
diff --git a/core/integration/tests/connectors/meilisearch/meilisearch_sink.rs
b/core/integration/tests/connectors/meilisearch/meilisearch_sink.rs
new file mode 100644
index 000000000..17a81db44
--- /dev/null
+++ b/core/integration/tests/connectors/meilisearch/meilisearch_sink.rs
@@ -0,0 +1,81 @@
+// 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.
+
+use crate::connectors::fixtures::{MeilisearchOps, MeilisearchSinkFixture,
TEST_INDEX};
+use bytes::Bytes;
+use iggy::prelude::{IggyMessage, Partitioning};
+use iggy_common::{Identifier, MessageClient};
+use integration::harness::seeds;
+use integration::iggy_harness;
+
+#[iggy_harness(
+ server(connectors_runtime(config_path =
"tests/connectors/meilisearch/sink.toml")),
+ seed = seeds::connector_stream
+)]
+async fn given_json_messages_when_sink_consumes_should_index_documents(
+ harness: &TestHarness,
+ fixture: MeilisearchSinkFixture,
+) {
+ let client = harness.root_client().await.unwrap();
+ let stream_id: Identifier = seeds::names::STREAM.try_into().unwrap();
+ let topic_id: Identifier = seeds::names::TOPIC.try_into().unwrap();
+
+ let payloads = [
+ serde_json::json!({"name": "first", "category": "alpha"}),
+ serde_json::json!({"name": "second", "category": "beta"}),
+ ];
+
+ let mut messages = payloads
+ .iter()
+ .enumerate()
+ .map(|(i, payload)| {
+ IggyMessage::builder()
+ .id((i + 1) as u128)
+
.payload(Bytes::from(serde_json::to_vec(payload).expect("serialize")))
+ .build()
+ .expect("build message")
+ })
+ .collect::<Vec<_>>();
+
+ client
+ .send_messages(
+ &stream_id,
+ &topic_id,
+ &Partitioning::partition_id(0),
+ &mut messages,
+ )
+ .await
+ .expect("send messages");
+
+ let documents = fixture
+ .wait_for_documents(TEST_INDEX, payloads.len())
+ .await
+ .expect("wait for Meilisearch documents");
+
+ assert_eq!(documents.len(), payloads.len());
+ assert!(documents.iter().any(|document| document["name"] == "first"));
+ assert!(
+ documents
+ .iter()
+ .any(|document| document["name"] == "second")
+ );
+ assert!(
+ documents
+ .iter()
+ .all(|document| document["iggy_id"].as_str().is_some())
+ );
+}
diff --git a/core/integration/tests/connectors/meilisearch/mod.rs
b/core/integration/tests/connectors/meilisearch/mod.rs
new file mode 100644
index 000000000..fd50dc367
--- /dev/null
+++ b/core/integration/tests/connectors/meilisearch/mod.rs
@@ -0,0 +1,18 @@
+// 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.
+
+mod meilisearch_sink;
diff --git a/core/integration/tests/connectors/meilisearch/sink.toml
b/core/integration/tests/connectors/meilisearch/sink.toml
new file mode 100644
index 000000000..0a846ac96
--- /dev/null
+++ b/core/integration/tests/connectors/meilisearch/sink.toml
@@ -0,0 +1,20 @@
+# 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.
+
+[connectors]
+config_type = "local"
+config_dir = "../connectors/sinks/meilisearch_sink"
diff --git a/core/integration/tests/connectors/mod.rs
b/core/integration/tests/connectors/mod.rs
index a1433160b..08b794fe8 100644
--- a/core/integration/tests/connectors/mod.rs
+++ b/core/integration/tests/connectors/mod.rs
@@ -25,6 +25,7 @@ mod http;
mod http_config_provider;
mod iceberg;
mod influxdb;
+mod meilisearch;
mod mongodb;
mod postgres;
mod quickwit;
