Connector Sdk

Extending the Apache Iggy Connectors SDK

The SDK (core/connectors/sdk/) is the stable contract between the runtime and every plugin. Changes ripple to every sink/source in-tree and every third-party plugin. Treat the public surface as a versioned API.

Universal connector rules (benchmark, SecretString, drop accounting, exemplar patterns) live in connectors-overview. This skill covers SDK-level changes only.

Contents

• STOP and ask the user before
• File map
• Cardinal rules
• Adding a new Schema variant
• Adding a StreamDecoder / StreamEncoder
• Adding a new Error variant
• Modifying Sink / Source / Transform traits
• FFI macros (sink_connector!, source_connector!)
• Payload::try_to_bytes(&self) is non-negotiable for JSON
• Retry helpers (retry.rs)
• ConnectorState
• Tests for SDK changes
• Before declaring done

STOP and ask the user before

• Bumping the SDK MAJOR version or changing any #[repr(C)] layout - misaligns every pre-built plugin .so.
• Changing Sink / Source / Transform / StreamDecoder / StreamEncoder trait signatures - cascades through every plugin.
• Removing or renaming a Schema variant or Error variant - decoders/encoders pinned to it. downstream pattern matches break.
• Changing wire serialization (postcard / rmp_serde / serde_json) for any FFI / state / config payload.
• Changing FFI macro return codes or function names (iggy_*_open / consume / handle / close) - runtime + plugins both need lockstep update.

File map

sdk/src/
├── lib.rs              Traits (Sink, Source, StreamDecoder, StreamEncoder),
│                       Payload, Schema, ConnectorState, Error enum, FFI message structs.
├── sink.rs             SinkContainer + sink_connector! macro (FFI plumbing).
├── source.rs           SourceContainer + source_connector! macro (FFI plumbing).
├── api.rs              ConnectorStatus, ConnectorStats (feature = "api").
├── convert.rs          owned_value_to_serde_json (simd_json ⇄ serde_json bridge).
├── log.rs              CallbackLayer for tracing across FFI.
├── retry.rs            retry_async + RetryPolicy, CircuitBreaker, HttpRetryMiddleware.
├── decoders/           One per schema: json, raw, text, proto, flatbuffer, avro.
├── encoders/           Mirror of decoders.
└── transforms/         add_fields, delete_fields, update_fields, filter_fields,
                        unwrap_envelope, proto_convert, flatbuffer_convert, avro_convert.

Cardinal rules

1. Apache 2.0 license header on every new file.
2. Send + Sync on every public trait (FFI runs across thread pools).
3. #[repr(C)] on every type that crosses FFI (Schema, TopicMetadata, MessagesMetadata, RawMessage, ProducedMessages, ConsumedMessage, DecodedMessage, etc. in lib.rs). Adding fields requires bumping the SDK version - existing plugins built against the old layout will misalign on postcard::from_bytes.
4. postcard for FFI message serialization (handled by SinkContainer::consume, SourceContainer's handle_messages). MessagePack (rmp_serde) for ConnectorState. JSON (serde_json) only for human-editable config that crosses FFI. Don't mix.
5. simd_json::OwnedValue for JSON payloads, not serde_json::Value. Use convert::owned_value_to_serde_json as a bridge when interop is required.
6. BTreeMap for headers - deterministic ordering. Never HashMap on the wire.
7. No breaking changes to Sink/Source/StreamDecoder/StreamEncoder/Transform trait signatures without coordinating with all in-tree plugins in the same PR.

Adding a new Schema variant

The Schema enum (in lib.rs) is #[repr(C)] - it crosses FFI. Touch points to update in one PR:

1. Schema enum: add variant with #[strum(to_string = "...")] matching the serde(rename_all="snake_case") form.
2. Payload enum: add a matching variant carrying the deserialized form.
3. Payload::try_into_vec - consuming bytes-out path.
4. Payload::try_to_bytes - borrowing bytes-out path. For non-trivial payloads (parsed trees), implement a no-clone serialization, not a clone() + serialize. See the Payload::Json arm for the canonical optimization.
5. Payload::Display.
6. Schema::try_into_payload - bytes → Payload.
7. Schema::decoder() - factory returning Arc<dyn StreamDecoder>.
8. Schema::encoder() - factory returning Arc<dyn StreamEncoder>.
9. New decoders/<name>.rs and encoders/<name>.rs.
10. Update sdk/README.md and core/connectors/README.md schema list.
11. Tests: round-trip encode/decode, error paths.

Miss any of these → silent failures at runtime (typically Error::InvalidPayloadType or surprising decoder behavior in plugin code).

Adding a StreamDecoder / StreamEncoder

Pattern from decoders/json.rs::JsonStreamDecoder:

pub struct MyStreamDecoder;        // unit struct if stateless

impl StreamDecoder for MyStreamDecoder {
    fn schema(&self) -> Schema { Schema::MyFormat }
    fn decode(&self, payload: Vec<u8>) -> Result<Payload, Error> {
        // Decode payload bytes into Payload::MyFormat(...).
        // Errors: Error::CannotDecode(Schema::MyFormat), Error::InvalidJsonPayload, etc.
    }
}

For stateful decoders (proto/avro/flatbuffer) holding schema descriptors:

• Store the schema state directly in the struct (config, message_descriptor, schema). The instance is built once and wrapped in Arc<dyn> for sharing - it must be Send + Sync but does not need additional locking if construction-time loading is final.
• Provide two constructors when schema loading can fail: lenient new(config) -> Self (logs + degrades) AND strict try_new(config) -> Result<Self, Error> (fail-fast). Plus a Default impl for the no-schema case. Canonical example: sdk/src/decoders/avro.rs exports all three (AvroStreamDecoder::new, ::try_new, Default).
• If you genuinely need to mutate the schema after construction (e.g., a update_config method), use &mut self plus std::mem::replace(&mut self.config, new_config) to swap without cloning the old value. Live pattern: sdk/src/decoders/avro.rs::AvroStreamDecoder::update_config and sdk/src/encoders/avro.rs::AvroStreamEncoder::update_config.
• A fresh decoder instance is created via Schema::decoder() on each call (it returns Arc<dyn StreamDecoder>), so per-decoder caching of shared mutable state is the wrong abstraction.

Adding a new Error variant

The Error enum (in lib.rs) is Clone + PartialEq + Eq + Hash. New variants must preserve these. Guidelines:

• Use existing variants first. Only add a new one if the failure mode is distinct in handling, not just description. InvalidConfigValue(String) covers most "bad config" cases.
• Carry context as String - don't add a struct unless multiple discrete fields are needed by callers programmatically.
• Document retry semantics in the docstring. Existing variants (PermanentHttpError, CatalogCommitError, TransactionApplyError) document retry behavior - mirror that style.
• Place near related variants to keep the enum readable top-to-bottom.

Modifying Sink / Source / Transform traits

Breaking changes. Required process:

1. Document migration in the SDK changelog (and connectors/sdk/README.md).
2. Update every in-tree plugin in the same PR. CI builds them all.
3. Bump SDK minor version.
4. Note in PR description: dynamic plugins compiled against the old SDK version will fail to load with mismatched FFI symbols.

Non-breaking additions (new default method, new struct field with #[serde(default)], new variant on a non-exhaustive enum) are preferable.

FFI macros (sink_connector!, source_connector!)

The macros (in sink.rs and source.rs) generate:

• iggy_<kind>_open(id, config_ptr, config_len, [state_ptr, state_len,] log_callback) -> i32
• iggy_sink_consume(id, ...) / iggy_source_handle(id, callback) -> i32
• iggy_<kind>_close(id) -> i32
• iggy_<kind>_version() -> *const c_char (static lifetime, from env!("CARGO_PKG_VERSION"))

Invariants:

• Duplicate-ID guard returns -1 if the caller reopens an ID without closing first. Don't remove this - it prevents silent buffered-data loss.
• INSTANCES: Lazy<DashMap<u32, SinkContainer<$type>>> (and SourceContainer<$type> mirror) is the only mutable global the macro introduces - keep it that way.
• Return codes: 0 success, -1 invalid call, 1 open failure. Don't repurpose.

Changes to the FFI signature must update runtime/src/main.rs::{SourceApi, SinkApi} in the same PR.

Payload::try_to_bytes(&self) is non-negotiable for JSON

Plugin authors call this on every consumed message. The implementation in lib.rs::Payload::try_to_bytes documents why it skips the deep OwnedValue clone - replacing O(n) clone + O(n) serialize with O(n) serialize. If you add a new variant requiring a parsed tree, do the same: serialize in place, don't clone.

Retry helpers (retry.rs)

• CircuitBreaker: threshold + cooldown, try_lock() on the success path to avoid hot-path contention.
• retry_async(policy, context, should_retry, op): the retry loop for anything failing as Err. Owns attempt counting, backoff and the per-retry log; returns RetryFailure { error, attempts, exhausted } so callers write their own terminal log.
• retry_backoff(base, retry, max): backoff only, for a loop that computes its own delay. retry is 1-based. exponential_backoff is the 0-based primitive underneath and applies no jitter, so call it directly only when the delay must be exact (source.rs::nack_retry_delay, whose tests assert exact values).
• HttpRetryMiddleware: integrates with reqwest-middleware. Retries 429 + 5xx + network errors. Honors Retry-After.
• max_retries = total attempts including the first try, not extra retries. Document if you change this convention. meilisearch_sink is the standing exception: its max_retries / max_open_retries count retries after the first, as its README states.
• New helpers must take Duration (not u64 millis) on the public API. Internal computation uses humantime parsing of String.

ConnectorState

• Bytes are opaque - the wrapping type does not impose schema.
• Serialization is MessagePack via rmp_serde. Compact, deterministic, well-supported in serde.
• serialize/deserialize helpers return Option<T> and log on failure. Failures are non-fatal - design downstream code to tolerate fresh state.
• Don't change the underlying format without coordinating with all sources.

Tests for SDK changes

• Unit tests for any new pure function in the changed module.
• Round-trip tests for new schemas/transforms (encode → decode == identity for JSON-equivalent payloads).
• sdk/tests/ integration tests when the change crosses module boundaries.

Before declaring done

cargo fmt --all
cargo sort --no-format --workspace
cargo clippy -p iggy_connector_sdk --all-targets --all-features -- -D warnings
cargo test -p iggy_connector_sdk --all-features

# Rebuild all plugins to catch breaking-change leaks:
cargo build -p iggy_connector_stdout_sink -p iggy_connector_random_source
# If FFI signatures changed, also build the runtime:
cargo build -p iggy-connectors

Discussion / help: see AGENTS.md.