# SDK

> The Sink and Source traits, the connector macros, and the decoders and encoders used to read and write Iggy streams.

Rendered page: https://iggy.apache.org/docs/connectors/sdk/

Source: https://github.com/apache/iggy-website/blob/main/content/docs/connectors/sdk.mdx

Docs version: server 0.9.0, Rust SDK 0.11.0, CLI 0.14.0, Node.js SDK 0.10.0, and the Python, Java, Go and C# SDKs at 0.9.0

SDK provides the commonly used structs and traits such as `Sink` and `Source`, along with the `sink_connector` and `source_connector` macros to be used when developing connectors.

Moreover, it contains both, the `decoders` and `encoders` modules, implementing either `StreamDecoder` or `StreamEncoder` traits, which are used when consuming or producing data from/to Iggy streams. All the supported schemas - **JSON**, **Raw bytes**, **Text**, **Protocol Buffers**, **FlatBuffers**, and **Avro** - have both a decoder and an encoder.

## Traits

A source produces messages to the configured stream and topic. A sink consumes messages from the configured stream(s) and topic(s):

```rust
use async_trait::async_trait;
use iggy_connector_sdk::{
    ConsumedMessage, Error, MessagesMetadata, ProducedMessages, TopicMetadata,
    source::SourceBatchResult,
};

#[async_trait]
pub trait Source: Send + Sync {
    async fn open(&mut self) -> Result<(), Error>;
    async fn poll(&self) -> Result<ProducedMessages, Error>;
    async fn on_batch_result(&self, _result: SourceBatchResult) -> Result<(), Error> {
        Ok(())
    }
    async fn close(&mut self) -> Result<(), Error>;
}

#[async_trait]
pub trait Sink: Send + Sync {
    async fn open(&mut self) -> Result<(), Error>;
    async fn consume(
        &self,
        topic_metadata: &TopicMetadata,
        messages_metadata: MessagesMetadata,
        messages: Vec<ConsumedMessage>,
    ) -> Result<(), Error>;
    async fn close(&mut self) -> Result<(), Error>;
}
```

Expose the implementation with `sink_connector!(MySink)` or `source_connector!(MySource)`. The sink constructor has the signature `new(id: u32, config: MySinkConfig) -> Self`; the source constructor also takes `state: Option<ConnectorState>`. The **[sink guide](https://iggy.apache.org/docs/connectors/sinks/sink)** and **[source guide](https://iggy.apache.org/docs/connectors/sources/source)** provide complete plugin implementations and build setup.

The macros generate the C FFI symbols loaded by the runtime: `iggy_sink_open`, `iggy_sink_consume`, `iggy_sink_close`, and `iggy_sink_version` for sinks, and `iggy_source_open`, `iggy_source_handle_v2`, `iggy_source_batch_result`, `iggy_source_close`, and `iggy_source_version` for sources. Configurations cross FFI as JSON, message batches and metadata use postcard, and source checkpoint bytes are opaque. The SDK creates one Tokio runtime per loaded plugin library, shared by that library's connector instances.

## State

Source connectors can persist their position between restarts. Each batch of `ProducedMessages` may carry an optional `ConnectorState`. After processing and sending the batch to Iggy, the runtime saves that state through the configured file or HTTP backend before acknowledging the batch. The last saved state is passed to `new()` on the next startup. `ConnectorState` wraps raw bytes and provides `serialize`/`deserialize` helpers backed by MessagePack.

Sources with cursors or destructive operations must stage changes during `poll()` and apply them on `SourceBatchResult::Ack`; `Nack` discards staged progress so the batch can be polled again. The SDK allows one batch in flight and stops polling if `on_batch_result()` returns an error. Its default no-op is suitable only for sources without staged work.

Sinks can keep in-memory state, but the runtime does not persist it. Their consumer-group offsets are tracked by Iggy. See the **[Observability page](https://iggy.apache.org/docs/connectors/observability)** for checkpoint storage and the **[sink guide](https://iggy.apache.org/docs/connectors/sinks/sink)** for offset handling.

## Retry utilities

The `retry` module ships helpers for connectors talking to flaky external systems: a `CircuitBreaker`, an `HttpRetryMiddleware` for `reqwest`-based clients, and a `build_retry_client` factory that wraps a reqwest client with the retry middleware (the circuit breaker is wired separately).

## Transforms

Last but not least, the different `transforms` are available, to transform (add, update, delete etc.) the particular fields of the data being processed via external configuration. It's as simple as adding a new transform to the `transforms` section of the particular connector configuration file:

```toml
[transforms.add_fields]
enabled = true

[[transforms.add_fields.fields]]
key = "message"
value.static = "hello"
```

All the available transforms and their configuration options are documented on the **[Transforms page](https://iggy.apache.org/docs/connectors/transforms)**.

## Protocol Buffers Support

The SDK includes support for Protocol Buffers (protobuf) format with both encoding and decoding capabilities. Protocol Buffers provide efficient serialization and are particularly useful for high-performance data streaming scenarios.

### Configuration Example

This example uses the Random source and Stdout sink from the matching checkout. Start a server using the **[getting-started guide](https://iggy.apache.org/docs/introduction/getting-started)**, then build the plugins, runtime, and CLI from the repository root:

```bash
cargo build --release -p iggy_connector_random_source -p iggy_connector_stdout_sink -p iggy-connectors -p iggy-cli
mkdir -p connectors
```

The source's `schema = "proto"` selects the default protobuf encoder, which wraps each JSON record in a `google.protobuf.StringValue` inside `google.protobuf.Any`. The sink reads raw bytes and applies `proto_convert` to expose the Any envelope as JSON. This does not unpack a custom protobuf message schema.

**Main runtime config (connectors.toml):**

```toml
[iggy]
address = "localhost:8090"
username = "iggy"
password = "iggy"

[connectors]
config_type = "local"
config_dir = "connectors"
```

**Source connector config (connectors/protobuf_source.toml):**

```toml
type = "source"
key = "protobuf"
enabled = true
version = 0
name = "Protobuf Source"
path = "target/release/libiggy_connector_random_source"

[[streams]]
stream = "protobuf_stream"
topic = "protobuf_topic"
schema = "proto"
batch_length = 1000
linger_time = "5ms"

[plugin_config]
interval = "100ms"
messages_range = [1, 10]
payload_size = 32
max_count = 100
```

**Sink connector config (connectors/protobuf_sink.toml):**

```toml
type = "sink"
key = "protobuf"
enabled = true
version = 0
name = "Protobuf Sink"
path = "target/release/libiggy_connector_stdout_sink"

[[streams]]
stream = "protobuf_stream"
topics = ["protobuf_topic"]
schema = "raw"

[plugin_config]
print_payload = true

[transforms.proto_convert]
enabled = true
source_format = "proto"
target_format = "json"
include_paths = ["."]
preserve_unknown_fields = false

[transforms.proto_convert.conversion_options]
validate_messages = true
pretty_json = false
include_metadata = false
type_url_prefix = "type.googleapis.com"
strict_mode = false
```

Create the stream and topic, then start the runtime from the repository root:

```bash
./target/release/iggy --username iggy --password iggy stream create protobuf_stream
./target/release/iggy --username iggy --password iggy topic create protobuf_stream protobuf_topic 1 none 1d
IGGY_CONNECTORS_CONFIG_PATH=connectors.toml ./target/release/iggy-connectors
```

The source sends 100 records, then continues polling without new messages. Stdout logs message offsets and the serialized JSON envelope bytes, containing `type_url` and base64 `value`. The sink's `raw` schema also determines how the plugin receives those transformed bytes.

The format-conversion transforms define no per-key defaults. Every non-optional key shown above must be present, or the configuration fails to deserialize (`schema_path`, `message_type`, `field_mappings`, and `descriptor_set` are optional).

The two `[[streams]]` shapes differ: a source produces to a single `topic` and can tune batching via `batch_length` and `linger_time`, while a sink consumes from a list of `topics` and can additionally set `batch_length`, `poll_interval`, and `consumer_group`.

### Key Configuration Options

#### Programmatic Encoder and Decoder Configuration

These are SDK configuration fields, not Random or Stdout `plugin_config` keys. The runtime's `schema = "proto"` uses the default encoder or decoder.

- **`schema_path`**: Path to the `.proto` file containing message definitions
- **`message_type`**: Fully qualified name of the protobuf message type to use
- **`use_any_wrapper`**: Selects the Any fallback when no message descriptor is loaded; a loaded descriptor takes precedence

#### Transform Options

- **`proto_convert`**: Transform for converting between protobuf and other formats
- **`source_format`** / **`target_format`**: Formats to convert between - any schema value (`json`, `raw`, `text`, `proto`, `flat_buffer`, `avro`)
- **`preserve_unknown_fields`**: Accepted by `proto_convert`, but currently has no effect
- **`include_paths`**: Additional directories searched for imported `.proto` files
- **`field_mappings`**: Renames fields in a JSON input object before conversion (e.g., `"old_field" = "new_field"`)
- **`conversion_options`**: `pretty_json` controls JSON text output and `include_metadata` enriches supported protobuf-to-JSON paths. `validate_messages`, `type_url_prefix`, and `strict_mode` are accepted but currently have no effect

The `schema_registry_url` field is reserved and currently not implemented. The SDK never contacts a schema registry, and schemas are loaded only from `schema_path` or `descriptor_set`.

### Supported Features

- **Encoding**: A loaded message descriptor encodes matching JSON fields. The encoder supports booleans, strings, all protobuf integer types, and base64 strings for bytes or already-encoded nested messages. Float, double, and enum fields are unsupported by the encoder; nested JSON objects, repeated fields, maps, and proto2 groups are not a general-purpose schema conversion path.
- **Decoding**: A loaded descriptor extracts present fields. Integer, boolean, and string fields become JSON values; bytes become base64 and nested messages become metadata with base64 content. Missing fields are not filled with protobuf defaults, and float/double fields produce `unsupported_wire_type` placeholders. Without a descriptor, the default decoder returns an Any envelope's `type_url` and base64 `value`.
- **Transforms**: `proto_convert` supports JSON-to-protobuf schema encoding for scalar fields, including floating-point numbers and numeric enum values; bytes and already-encoded nested messages use base64 strings. It logs and omits fields it cannot encode. Its protobuf-to-JSON path exposes Any metadata or raw-data metadata rather than decoding a custom message descriptor. Converting protobuf to `flat_buffer` or `avro` rewraps bytes without transcoding them.
- **Field Mapping**: Encoder/decoder mappings use protobuf field names as keys and JSON field names as values. The encoder applies that mapping in reverse. Transform mappings rename JSON input keys directly.
- **Any Wrapper**: The default encoder puts JSON/text in a `google.protobuf.StringValue`, or binary data in a `google.protobuf.BytesValue`, inside `google.protobuf.Any`. The default decoder exposes the envelope without unpacking its inner message.

### Programmatic Usage

From the matching repository root, create an example crate and schema directory:

```bash
mkdir -p connector-sdk-example/src schemas
```

Save this as `connector-sdk-example/Cargo.toml`. The path dependency uses the SDK from the same checkout as the runtime:

```toml
[package]
name = "connector-sdk-example"
version = "0.1.0"
edition = "2024"

[dependencies]
iggy_connector_sdk = { path = "../core/connectors/sdk" }
simd-json = { version = "0.18.1", features = ["serde_impl"] }

[workspace]
```

Save this as `schemas/user.proto`:

```protobuf
syntax = "proto3";
package com.example;

message User {
  uint64 id = 1;
  string name = 2;
}
```

Each Rust example below is a complete `connector-sdk-example/src/main.rs`. Run it from the repository root so the relative schema path resolves:

```bash
cargo run --manifest-path connector-sdk-example/Cargo.toml
```

#### Dynamic Schema Loading

You can load or reload schemas programmatically:

```rust
use iggy_connector_sdk::decoders::proto::{ProtoConfig, ProtoStreamDecoder};
use iggy_connector_sdk::encoders::proto::{ProtoEncoderConfig, ProtoStreamEncoder};
use iggy_connector_sdk::{Error, Payload, StreamDecoder, StreamEncoder};
use std::path::PathBuf;

fn main() -> Result<(), Error> {
    let mut decoder = ProtoStreamDecoder::new_default();
    decoder.update_config(
        ProtoConfig {
            schema_path: Some(PathBuf::from("schemas/user.proto")),
            message_type: Some("com.example.User".to_string()),
            ..ProtoConfig::default()
        },
        true,
    )?;
    let encoder = ProtoStreamEncoder::new_with_config(ProtoEncoderConfig {
        schema_path: Some(PathBuf::from("schemas/user.proto")),
        message_type: Some("com.example.User".to_string()),
        ..ProtoEncoderConfig::default()
    });
    let encoded = encoder.encode(Payload::Json(simd_json::json!({
        "id": 1,
        "name": "Alice"
    })))?;
    println!("{}", decoder.decode(encoded)?);
    Ok(())
}
```

The encoder follows the same pattern:

```rust
use iggy_connector_sdk::encoders::proto::{ProtoEncoderConfig, ProtoStreamEncoder};
use iggy_connector_sdk::{Error, Payload, StreamEncoder};
use std::path::PathBuf;

fn main() -> Result<(), Error> {
    let mut encoder = ProtoStreamEncoder::new_with_config(ProtoEncoderConfig {
        schema_path: Some(PathBuf::from("schemas/user.proto")),
        message_type: Some("com.example.User".to_string()),
        use_any_wrapper: false,
        ..ProtoEncoderConfig::default()
    });
    encoder.load_schema()?;
    let encoded = encoder.encode(Payload::Json(simd_json::json!({
        "id": 1,
        "name": "Alice"
    })))?;
    println!("{encoded:?}");
    Ok(())
}
```

#### Creating Converters with Schema

The loaded schema is used for JSON-to-protobuf conversion. This example maps `user_id` and `full_name` to the schema's field names:

```rust
use iggy_connector_sdk::transforms::{ProtoConvert, ProtoConvertConfig, Transform};
use iggy_connector_sdk::{DecodedMessage, Error, Payload, Schema, TopicMetadata};
use std::collections::HashMap;
use std::path::PathBuf;

fn main() -> Result<(), Error> {
    let converter = ProtoConvert::new(ProtoConvertConfig {
        source_format: Schema::Json,
        target_format: Schema::Proto,
        schema_path: Some(PathBuf::from("schemas/user.proto")),
        message_type: Some("com.example.User".to_string()),
        field_mappings: Some(HashMap::from([
            ("user_id".to_string(), "id".to_string()),
            ("full_name".to_string(), "name".to_string()),
        ])),
        ..ProtoConvertConfig::default()
    });
    let metadata = TopicMetadata {
        stream: "users".to_string(),
        topic: "users".to_string(),
    };
    let message = DecodedMessage {
        id: None,
        offset: None,
        checksum: None,
        timestamp: None,
        origin_timestamp: None,
        headers: None,
        payload: Payload::Json(simd_json::json!({
            "user_id": 1,
            "full_name": "Alice"
        })),
    };
    if let Some(converted) = converter.transform(&metadata, message)? {
        println!("{:?}", converted.payload);
    }
    Ok(())
}
```

### Usage Notes

- **Automatic Loading**: Constructors attempt to load `schema_path` or `descriptor_set`; `schema_path` takes precedence when both are set. Constructors log loading errors and return an instance without a loaded schema.
- **Manual Loading**: `load_schema()` reloads the configured source. A returned error preserves an already-loaded schema. `update_config(config, true)` also restores the previous configuration on error; `false` changes the configuration while retaining the cached schema.
- **Fallbacks**: Missing files, compilation failures, absent schema sources, or an unmatched `message_type` can return `Ok(())` without an active message descriptor. Successful reloads into fallback mode clear the previous descriptor. Invalid protobuf syntax and malformed descriptor bytes return errors. Check the actual encoded/decoded result when validating a schema setup.
- **Encoding Errors**: Errors encoding a loaded message descriptor are returned to the caller. The encoder does not retry that message as Any. The decoder attempts Any after a schema decoding error.
- **Transform Configuration**: Create a new converter to change its configuration. `load_schema()` can reload its existing source. Without a descriptor, JSON-to-protobuf conversion produces JSON text in `Payload::Proto`, not a schema-encoded binary message.
- **Format Options**: Encoder `preserve_unknown_fields`, `compact_encoding`, `validate_message`, and `deterministic_encoding` are accepted but have no effect. Decoder `preserve_unknown_fields` retains unknown varints as numbers and length-delimited data as base64; fixed-width unknown fields become placeholders. It does not retain the original wire encoding. See the [Transforms page](https://iggy.apache.org/docs/connectors/transforms) for conversion-specific limits.
- Protocol Buffers provide efficient binary serialization compared to JSON