Connector Source

Writing an Apache Iggy Connector Source

A source is a Rust cdylib that implements iggy_connector_sdk::Source and exposes FFI symbols via the source_connector! macro. The runtime calls poll() in a loop, applies transforms, encodes via the configured Schema, sends to Apache Iggy, and persists the returned ConnectorState after every successful send.

Universal connector rules (SecretString, benchmark, verbose flag, drop accounting, filter contract, exemplar patterns) live in connectors-overview. This skill covers only what's source-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 (Source::open / poll / on_batch_result / close) - that's an SDK change.
• Adding a long-running side task in the plugin - the runtime owns lifecycle, and orphans survive close(). Sanctioned only where the source is itself a server the runtime cannot drive, as in http_source's listener, and then only with an explicit shutdown in the last close() that awaits its tasks before returning.
• Persisting unbounded state - State is rewritten every batch.
• Adding a source that requires authoritative offsets external to Apache Iggy without coordinating retention.

Quick reference

• Skeleton: TEMPLATE.md (load on demand).
• Exemplars: random_source (minimal + canonical state tests), postgres_source (cursor / delete-after-read / processed-column modes, restart-survives-state tests), elasticsearch_source (scroll cursor), influxdb_source (time-series scan).

Hard rules

poll() signature is &self

The macro shares the source as Arc<T> across the FFI callback and forwarding loop. Signature: async fn poll(&self) -> ... - any mutable state behind tokio::sync::Mutex. Single most common new-contributor mistake.

Lock discipline

Never hold the state Mutex across upstream I/O. Build a candidate from committed state, then stage it until the runtime reports the batch result:

let mut candidate = self.state.lock().await.clone();
let rows = client.query(&sql, &[&candidate.cursor]).await?;
candidate.cursor = Some(new_cursor);
let persisted = ConnectorState::serialize(&candidate, CONNECTOR_NAME, self.id)
    .ok_or_else(|| Error::Serialization("failed to serialize source state".into()))?;
*self.pending.lock().await = Some(candidate);

Delivery acknowledgment

on_batch_result (added by #3855) is how a source learns what happened to the batch it just returned. The SDK keeps exactly one batch in flight: it will not call poll() again until this returns, and it stops the source after MAX_CONSECUTIVE_NACKS (5) consecutive NACKs, roughly 1.5s of backoff, without calling close().

• Ack means the runtime sent the batch and persisted its state. Nack means it could not confirm both, which is not the same as neither happening: a batch that reached the topic but whose state save failed is NACKed, and the SDK NACKs on its own result timeout while the send may still have landed. A source that replays on Nack is at-least-once, not exactly-once.
• The trait has a default no-op, which suits only a source with no staged cursor and no destructive work. If poll() advances a cursor, deletes rows, or drains an in-memory buffer, omitting this loses data silently and nothing will tell you. random_source and http_source implement it; the other shipped sources take the default and skip rows on a NACK.
• Stage in poll(), apply on Ack, discard or replay on Nack. A source whose input is pushed to it, rather than re-readable upstream, has to hold the batch itself: see http_source's staging.
• Returning Err from it stops the source immediately, so it is not a retry signal.

State persistence

• ConnectorState is Vec<u8> via MessagePack (rmp_serde). Use ConnectorState::serialize(&state, NAME, id) and ConnectorState::deserialize::<State>(NAME, id).
• poll() must not commit cursors or destructive work. Return messages with candidate state and keep the corresponding work staged.
• The runtime sends the batch, saves its candidate state to {state_path}/source_{key}.state, then calls on_batch_result(Ack). Commit staged in-memory state and external delete or mark operations only on ACK. A NACK discards the candidate so the same data can be polled again.
• A crash between poll() returning and state persistence leaves the prior cursor for the next poll, so downstream must tolerate at-least-once delivery.
• Cursor sources that stage state until on_batch_result can attach state to the corresponding message batch. Control-plane state that is independent of a publish can ride an empty batch instead.
• Return state: None for an empty poll when no watermark changed. If an empty poll advances a watermark, stage and return the new state through the same ACK handshake.
• The runtime can still NACK an empty batch when state storage is latched or a pending checkpoint cannot resolve. Never treat hand-off as durable; on_batch_result reports the outcome.
• Treat candidate-state serialization failure as a poll error. Do not send messages without the state needed to resume them safely.
• Keep State small - rewritten every batch. No unbounded vecs.

The SDK allows one in-flight batch. Five consecutive NACKs stop the source and require a manual restart. Returning Err from on_batch_result is fatal, so retry transient backend failures inside the callback before returning an error. The runtime must report ACK or NACK within the SDK's 30-second batch-result window. Once the result is received, the SDK waits for on_batch_result to finish, so the callback must bound its own connection acquisition and retry budget rather than relying on the SDK deadline.

Sleep first

poll() must sleep(self.poll_interval).await before any work. Without it, an empty source spins a CPU.

Schema selection

Match ProducedMessages.schema to the bytes in messages[i].payload:

• JSON-serialized rows → Schema::Json
• Already-protobuf bytes → Schema::Proto
• Already-avro bytes → Schema::Avro
• Opaque → Schema::Raw

IDs and timestamps

• ProducedMessage.id: Option<u128> - set when a natural ID exists (DB PK, document id). It rides the wire as the message id, which is what a consumer can dedupe an at-least-once duplicate on.
• Iggy itself does not dedupe on that id. The server's only read of the field is core/server/src/http/wire.rs, which mints a fresh uuid when the incoming id is 0 and otherwise passes it through. Its dedup path is a per-client request-id watermark (dedup_clients_max) that never looks at this field.
• origin_timestamp: Option<u64> - source-system event time in nanoseconds. Lets downstream sinks reason about lag.
• timestamp and checksum are Iggy-side - leave None.

Concurrency

• Runtime spawns ONE poll() task per source. No concurrent poll().
• Don't spawn your own long-running Tokio tasks: the runtime owns lifecycle. The exception is a source that listens rather than polls, which has to own its listener; http_source is the worked example, and it shuts its tasks down in the last close() rather than leaving them to outlive the connector.

Errors

Returning Err from poll() is only logged by the SDK's FFI bridge (sdk/src/source.rs::handle_messages) - the loop continues, the next poll() runs. Connector status does NOT flip to Error from a poll failure. Status Error is set by the runtime only on transform/encode failure, Iggy send failure, or state save failure (runtime/src/source.rs::source_forwarding_loop calls to context.sources.set_error). To surface a poll failure as unhealth, raise it through the metric counter or escalate to Error::InitError from open().

Logging

info!("Opened <connector> connector ID: {}, endpoint: {}", self.id, ...);
info!("Restored state for <connector> ID: {id}, cursor: {:?}", ...);
debug!("Polled {} rows for <connector> ID: {}", rows.len(), self.id);
warn!("Transient fetch failure for <connector> ID: {}, will retry: {error}", self.id);
error!("Failed to <op> for <connector> ID: {}, error: {error}", self.id);
info!("Closed <connector> connector ID: {}, total produced: {}", self.id, ...);

Iggy consumer-loop labels use literal API names (offset=, current_offset=).

Common pitfalls

1. async fn poll(&mut self) - won't compile. Use &self + Mutex<State>.
2. Holding state.lock() across the fetch I/O - blocks close(), causes shutdown timeouts.
3. Forgetting to sleep - 100% CPU on idle source.
4. Committing a cursor or deleting source data in poll() - stage it and wait for ACK.
5. Unbounded data in State - rewritten every batch. keep O(constant).
6. std::sync::Mutex - blocks the executor. Use tokio::sync::Mutex.
7. Not setting ProducedMessage.id when a stable ID exists - leaves a consumer nothing to dedupe a replayed duplicate on. It does not make the write idempotent server-side, because nothing there reads it.
8. Spawning side tasks - the runtime owns the scheduler. The one exception is a source that listens rather than polls and must own its listener (see Concurrency and the STOP list); it owes an explicit shutdown in the last close() that awaits its tasks.

Tests

Mandatory four canonical source state tests (see connector-testing for the full pattern). Copy from sources/random_source/src/lib.rs::tests. Plus config defaults, payload building, schema selection.

Integration tests under core/integration/tests/connectors/<backend>/ for any source backed by external infra. Use #[iggy_harness] + a TestFixture backed by testcontainers-modules. Reference: core/integration/tests/connectors/postgres/postgres_source.rs (multi-mode tests) + restart.rs (state survives restart). Exercise both ACK and NACK paths when the source stages cursors or destructive work.

Before declaring done

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

# Integration tests:
cargo test -p integration -- connectors::<backend>::<test_name>

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

Discussion / help: see AGENTS.md.