Connector Sink

Writing an Apache Iggy Connector Sink

A sink is a Rust cdylib that implements iggy_connector_sdk::Sink and exposes FFI symbols via the sink_connector! macro. The runtime loads it as .so/.dylib/.dll, polls Apache Iggy on its behalf, runs the transform chain, decodes via the configured Schema, and hands the sink batches of ConsumedMessage. The sink is responsible for getting them to the external system reliably and efficiently.

Universal connector rules (SecretString, benchmark, verbose flag, drop accounting, filter contract, exemplar patterns) live in connectors-overview. This skill covers only what's sink-specific.

Contents

• STOP and ask the user before
• Quick reference
• Hard rules
• Common pitfalls
• Tests
• Before declaring done

STOP and ask the user before

• Changing the SDK trait surface (Sink::open / consume / close) - that's an SDK change, not a plugin one.
• Touching default batch_size, retry caps, or backoff defaults baked into existing in-tree sinks.
• Promoting an Err variant to retryable when it previously was not - retrying bad data trips circuit breakers.
• Adding a new sink that introduces a backend the runtime doesn't yet integration-test for.

Quick reference

• Skeleton: TEMPLATE.md (load on demand when authoring).
• Exemplars: stdout_sink (minimal), postgres_sink (DB + transient detection), http_sink (validation, batch modes, retry middleware), mongodb_sink (atomic counters), elasticsearch_sink / iceberg_sink (backend-specific idioms).

Hard rules

Lifecycle

• Sink::consume takes &self - never &mut self. Mutable state goes behind tokio::sync::Mutex / AtomicU64. Prefer atomics for hot counters - mongodb_sink uses messages_processed: AtomicU64 + insertion_errors: AtomicU64 and avoids a state lock on the hot path entirely.
• Sink::open builds the client AND tests connectivity. Fail fast with Error::InitError.
• Sink::close flushes + drops resources. Take ownership via client: Option<Client> + .take(). sqlx pools have .close().await. reqwest/mongodb/elasticsearch clients just drop. Log final stats either way.

Config

• Every runtime-tunable field is Option<T> with a default applied in new(). Forward-compat: adding a field doesn't break existing TOML.
• Durations: Option<String>, parsed via humantime::Duration in new(). Two equivalent in-tree idioms - inline HumanDuration::from_str(&raw).map(|d| *d).unwrap_or_else(..) (random_source, postgres_source) or a local parse_duration helper (http_sink, sdk/src/retry.rs). Fall back to a default on parse failure with warn! - never panic. We do not use humantime_serde.
• Conflict resolution: if two config fields can be in invalid relative ordering, fix it in new() + warn! - don't error. http_sink swaps retry_delay and max_retry_delay when reversed to avoid an ExponentialBackoff panic downstream.
• Validation order: structural in new(), connectivity in open(), never in consume().

Payloads (efficiency-critical)

• Dispatch on messages_metadata.schema and the Payload variant. Handle Json, Raw, Text at minimum. Unsupported -> Error::InvalidPayloadType or fall back to base64 (elasticsearch_sink).
• For Payload::Json, call try_to_bytes(&self) - single-pass serialization without cloning the OwnedValue tree (sdk/src/lib.rs::Payload::try_to_bytes). Never payload.clone().try_into_vec().
• To take a payload out of ConsumedMessage without cloning: std::mem::replace(&mut message.payload, Payload::Raw(vec![])). Standard pattern across http_sink::send_individual / send_ndjson / send_json_array.
• Pre-allocate per-batch buffers with Vec::with_capacity(messages.len()) (or * factor for serialized bodies). Used across postgres_sink, mongodb_sink, quickwit_sink, elasticsearch_sink, http_sink.

Headers

• message.headers: Option<BTreeMap<HeaderKey, HeaderValue>>. Check .is_empty() before iterating - runtime treats empty as None.
• For binary values use v.as_raw() + base64. for text use v.to_string_value(). See http_sink::EncodedHeader.
• BTreeMap for deterministic ordering. Don't collect into HashMap.

Errors

Retry

• SDK helpers cover the simple case: iggy_connector_sdk::retry::check_connectivity_with_retry(...) for open(), HttpRetryMiddleware for default 429/5xx/network policy on reqwest clients.
• Custom strategies: reqwest-middleware + reqwest_retry::RetryTransientMiddleware::new_with_policy_and_strategy. http_sink defines its own HttpSinkRetryStrategy (honors success_status_codes, per-status decisions).
• Non-HTTP clients: write is_transient_error(&e) mapping driver-specific codes. postgres_sink::is_transient_error maps SQLSTATEs 40001, 40P01, 57P01-03, 08000/03/06.
• Retry loop: new connectors use iggy_connector_sdk::retry::retry_async(policy, context, should_retry, op) for anything that fails as Err. It owns attempt counting, backoff and the per-retry log, and returns RetryFailure { error, attempts, exhausted }; the caller logs the terminal failure.
• Backoff only, for a loop that computes its own delay (it retries on an Ok response, carries a deadline, or reconnects between attempts): retry_backoff(base, retry, max), where retry is 1-based. exponential_backoff is the 0-based primitive underneath, and it applies no jitter; call it directly only when the delay must be exact, as sdk/src/source.rs::nack_retry_delay needs for its tests.
• Cap retries at 3 total attempts.

Idempotency

Runtime retries on any Err - same batch can be redelivered. In-tree guarantees:

• Postgres (postgres_sink::build_batch_insert_query): PK on id, but plain INSERT (no ON CONFLICT). Duplicate retry triggers 23505, is_transient_error does NOT mark it transient -> error propagates. Adding ON CONFLICT (id) DO NOTHING is a known improvement.
• MongoDB: composite _id from stream:topic:partition:message_id, via insert_many(docs).ordered(false). Per-document duplicate-key errors don't abort the batch - effective idempotency under retry.
• Elasticsearch (bulk_index_documents): bulk action specifies only _index, no _id -> ES auto-generates ids -> NOT idempotent. Extensions should add "_id".
• HTTP: endpoint-dependent. Document in the plugin README.
• Iceberg: transactional commit. Error::CatalogCommitError documents that the txn is consumed on commit(). Rebuild-then-retry can duplicate data - callers must check the catalog first.

When writing a new sink, prefer dedup-on-write using the message id field. Mongo's upsert-style is the cleanest reference.

Batching

• Default batch_size: 100. Tune per-backend.
• Chunk via messages.chunks(self.batch_size). Do not buffer across consume() - runtime already batches at the consumer-poll boundary.
• For HTTP backends support multiple batch modes when applicable (individual / ndjson / json_array / raw). See http_sink::BatchMode.

Logging

info!("Opened <connector> connector ID: {}, endpoint: <redacted>", self.id);
debug!("Processing batch of {} messages, offset: {}", batch.len(), offset);
warn!("Retry {attempt}/{max} for <connector> ID: {}, reason: {error}", self.id);
error!("Failed to <op> for <connector> ID: {}, error: {error}", self.id);
info!("Closed <connector> connector ID: {}, processed: {}", self.id, count);

Use literal API field names as labels (offset=, current_offset=).

Common pitfalls

1. Missing [lib] crate-type = ["cdylib", "lib"] - runtime can't dlopen a regular rlib.
2. Cloning the whole Payload::Json to serialize - use try_to_bytes(&self).
3. Holding Mutex across .await when the locked region does I/O.
4. &mut self on consume - won't compile. you need interior mutability.
5. Returning the first error and dropping subsequent batches - track last_err, process all, return.
6. Forgetting to map PermanentHttpError - retries hammer the backend with bad data.
7. Mutating config in consume() - config is a snapshot from new(). Reload requires runtime restart.
8. Logging the connection string. Redact at construction: http_sink::sanitize_url_for_log populates HttpSink.log_url.

Tests

Unit tests at EOF of src/lib.rs. See connector-testing for BDD naming + test_config() helper.

Real-infra integration tests live under core/integration/tests/connectors/<backend>/ using #[iggy_harness] + a backend fixture that spins up Docker via testcontainers-modules. Plugins integrating with external services must have an integration test. Reference: core/integration/tests/connectors/postgres/postgres_sink.rs + fixtures/postgres/.

Before declaring done

cargo fmt --all
cargo sort --no-format --workspace
cargo clippy -p iggy_connector_<name>_sink --all-targets -- -D warnings
cargo test -p iggy_connector_<name>_sink

# Integration tests (filter by name; integration crate has no per-area target):
cargo test -p integration -- connectors::<backend>::<test_name>

Update core/connectors/sinks/README.md and add a sample TOML under core/connectors/runtime/example_config/connectors/.

Discussion / help: see AGENTS.md.