Meilisearch Sink

The Meilisearch sink connector consumes messages from Iggy streams and indexes them as documents in a Meilisearch index through the official Rust SDK. It can create the index on startup and waits for Meilisearch indexing tasks to complete by default, so task failures surface as connector errors.

This page is a curated subset of the documentation. The canonical reference is the upstream meilisearch_sink README in the apache/iggy repository.

Configuration

Use the broker credentials and runtime setup from the sink guide. From the matching 0.9.0/edge repository root, build the plugin:

cargo build --release -p iggy_connector_meilisearch_sink

Save this connector file in the runtime's connector directory. Replace the URL and API key with your Meilisearch endpoint and credentials. The key needs permission to inspect the index, submit documents and read tasks, plus create the index when needed. Startup checks health, checks the index, and waits for index creation to succeed even when wait_for_tasks = false. Set primary_key to match an existing index. A mismatch only produces a startup warning; the connector does not change the index key, and document tasks can fail.

type = "sink"
key = "meilisearch"
enabled = true
version = 0
name = "Meilisearch sink"
path = "target/release/libiggy_connector_meilisearch_sink"
plugin_config_format = "toml"

[[streams]]
stream = "events"
topics = ["search_events"]
schema = "json"
batch_length = 1000
poll_interval = "5ms"
consumer_group = "meilisearch_sink"

[plugin_config]
url = "http://localhost:7700"
index = "iggy_messages"
api_key = "replace_with_secret_key"
primary_key = "iggy_id"
document_action = "replace"
batch_size = 1000

Create the Iggy resources and send a document:

./target/release/iggy --username iggy --password iggy stream create events
./target/release/iggy --username iggy --password iggy topic create events search_events 1 none 1d
./target/release/iggy --username iggy --password iggy message send --partition-id 0 events search_events '{"title":"First event","category":"example"}'

Common Options

Further options cover task polling (task_timeout 30s, task_poll_interval 100ms) and transient retries (max_retries 3, retry_delay 500ms, max_retry_delay 5s, max_open_retries 5). Both retry limits count retries after the first request, so max_retries = 3 permits up to four attempts within the operation deadline. Backoff doubles from retry_delay, adds ±20% jitter and is capped at max_retry_delay; reversed delay bounds are swapped. Invalid duration strings warn and fall back to 1s. There is no single deadline for the whole startup sequence. See the upstream meilisearch_sink README for details.

Document Mapping

JSON object payloads are indexed as documents directly. JSON arrays and scalars are wrapped in a value field, since Meilisearch documents must be objects. Raw payloads are parsed as JSON when possible and otherwise indexed in data as base64, with data_type = "raw" and data_encoding = "base64". Text payloads land in text with data_type = "text". These mappings apply after stream decoding and transforms; use schema = "raw" or schema = "text" for those representations. Unsupported payload variants are skipped with a warning and counted in the plugin's private error counter. A successful callback after these drops can still count the whole batch as processed in runtime statistics.

When the configured primary key is absent from a document, the connector injects a stable value derived from the stream, topic, partition, offset, and message ID, avoiding Meilisearch primary-key inference failures. An existing key is preserved, including an invalid or null value that Meilisearch may reject, unless a reserved metadata field below overwrites it. Use unique values that meet Meilisearch's document ID rules. Reusing a key with replace replaces the whole document, removing omitted fields; update keeps fields omitted from the new document. Both actions can insert a new document.

With include_metadata = true, stream, topic, partition, offset, checksum, message ID, timestamps and available headers are written as reserved iggy_* fields after payload parsing, overwriting same-named payload fields. The exception is a supplied iggy_id when it is the configured primary key. With a different primary key, iggy_id holds the generated Iggy identity. Avoid other reserved metadata fields as your primary key because metadata overwrites them.

iggy_checksum and iggy_message_id are strings; offsets and message timestamps remain numbers. Message timestamps are microseconds, while iggy_ingested_at uses current wall-clock milliseconds. Disabling metadata leaves payload fields intact and still injects a missing primary key.

Delivery Semantics

Each polled batch is split into requests of at most batch_size documents, with no accumulation across polls. The first failed chunk stops the loop; later chunks are not attempted. Earlier chunks may already be indexed.

Transient submission and task-status errors are retried within their budgets. A failed indexing task is returned as an error without resubmitting it. A task timeout does not cancel the remote task, so it may finish later. The runtime records plugin errors and continues polling with consumer auto-commit; it does not replay failed batches. This does not provide an end-to-end at-least-once guarantee, and retries or manual replay after uncertain outcomes can repeat writes.

Setting wait_for_tasks = false returns after submission, before Meilisearch confirms indexing. Later task failures are neither observed nor retried by the connector. Index-creation tasks are still awaited during startup. Closing the connector does not wait for outstanding document tasks.

Transforms

Transforms can be applied before indexing into Meilisearch. See the transforms documentation for the available types and their configuration.