SurrealDB Sink

The SurrealDB sink connector writes messages from Iggy streams into SurrealDB over the HTTP API. Each poll is split into chunks of at most batch_size messages, with one SurrealQL bulk INSERT IGNORE ... RETURN NONE for each chunk containing valid records. Every record gets a deterministic record id derived from stream, topic, partition, offset, and message id, so replaying the same identities leaves existing records untouched. Messages are not buffered across polls.

This page is a curated subset of the documentation. The canonical reference is the upstream surrealdb_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 Iggy checkout root, build the plugin:

cargo build --release -p iggy_connector_surrealdb_sink

Start a local SurrealDB instance with the credentials used below. This example stores SurrealDB data in memory:

docker run -d --name iggy-surrealdb -p 127.0.0.1:8000:8000 surrealdb/surrealdb:v3.1.4 start --bind 0.0.0.0:8000 --user root --pass root memory
curl --fail http://127.0.0.1:8000/health

After the health request succeeds, save this connector file in the runtime's connector directory. Its plugin path assumes the runtime starts from the checkout root.

type = "sink"
key = "surrealdb"
enabled = true
version = 0
name = "SurrealDB sink"
path = "target/release/libiggy_connector_surrealdb_sink"
plugin_config_format = "toml"

[[streams]]
stream = "example_stream"
topics = ["example_topic"]
schema = "json"
batch_length = 1000
poll_interval = "5ms"
consumer_group = "surrealdb_sink_connector"

[plugin_config]
endpoint = "127.0.0.1:8000"
namespace = "iggy"
database = "connectors"
table = "iggy_messages"
username = "root"
password = "root"
auto_define_table = true
batch_size = 1000

Create the Iggy resources and send a JSON message, then start the runtime as described in the sink guide:

./target/release/iggy --username iggy --password iggy stream create example_stream
./target/release/iggy --username iggy --password iggy topic create example_stream example_topic 1 none 1d
./target/release/iggy --username iggy --password iggy message send --partition-id 0 example_stream example_topic '{"event":"created","count":1}'

Query the stored records:

curl --fail --user root:root -H 'Accept: application/json' -H 'Surreal-NS: iggy' -H 'Surreal-DB: connectors' --data 'SELECT * FROM iggy_messages;' http://127.0.0.1:8000/sql

Common Options

The include_headers, include_checksum, and include_origin_timestamp flags also default to true and work independently of include_metadata. Retry and timeout options are query_timeout (30s per HTTP request), max_retries (3 total write attempts, with 0 raised to 1), retry_delay (100ms), and max_retry_delay (5s). Invalid duration strings log a warning and fall back to 1s; zero durations are accepted. The maximum delay is raised to the base delay if configured smaller. verbose_logging defaults to false and raises per-poll submission logs from debug to info. See the upstream surrealdb_sink README for the full table.

Namespace, database and table names must start with an ASCII letter or underscore and contain only ASCII letters, digits and underscores. Endpoint URLs cannot contain credentials, a path beyond /, a query or a fragment. An explicit http:// or https:// scheme takes precedence over use_tls. Payload-format and authentication-scope names are case-insensitive; unknown values prevent startup.

With root, namespace or database authentication, startup calls /signin and SQL requests use Basic authentication with the configured scope. Namespace/database users require pre-existing resources and auto_define_table = false. With auth_scope = "none", credentials are ignored; the server must permit those unauthenticated operations. Automatic DDL uses IF NOT EXISTS and does not migrate existing definitions. The optional <table>_iggy_offset_idx index is not a uniqueness constraint. define_indexes = true with metadata disabled rejects startup; without automatic DDL it only logs a warning and skips index creation.

Stored Shape

Every submitted record contains id, iggy_message_id, payload and payload_encoding. include_metadata adds iggy_stream, iggy_topic, iggy_partition_id, iggy_offset, iggy_timestamp and iggy_schema. The checksum and origin timestamp have separate flags. iggy_headers appears only when enabled and nonempty.

Iggy message IDs, partition IDs, offsets, timestamps and checksums are stored as strings. Timestamps retain Iggy's microsecond values, not SurrealDB datetime values. Record IDs encode stream/topic UTF-8 bytes as hex, then append partition, offset and the 32-digit hexadecimal message ID. Changing the payload or transforms does not change that identity; changing any identity component creates a different key.

Payload formatting operates after the runtime decoder and transforms:

Non-raw headers are stored as strings, including numbers and booleans. Raw headers use {"data":"AQID","iggy_header_encoding":"base64"}. Payload JSON is sent as SurrealQL values, so destination schema, numeric and other value constraints still apply.

Delivery Semantics

The runtime auto-commits while polling, logs/counts plugin callback failures, and continues without replaying the failed batch. The sink retries transaction-conflict errors, connection/timeouts, and HTTP 408, 429, 500, 502, 503 and 504. Delays use capped exponential backoff with jitter; Retry-After is not used. Connection errors trigger reconnection and repeat sign-in/health/optional DDL. A failed reconnect stops that chunk, and retries plus startup requests can exceed query_timeout in total.

A malformed record is skipped while valid records in its chunk are still submitted. A failed chunk does not stop later chunks, and the last error is returned. The sink checks HTTP and SQL statement statuses. Its own processed counter counts records submitted in successful statements, including records ignored by INSERT IGNORE; runtime counters instead reflect whether the whole callback succeeded. Neither is a count of newly inserted rows.

SurrealQL INSERT IGNORE preserves existing records with the same ID, including their old payloads. With the SurrealDB 3.1.4 instance above, IGNORE also silently skips unique-index conflicts and field-assertion failures while accepting valid records from the same chunk. These skips return a successful statement and are included in processed counters. This protects repeated writes of the same identities; it does not ensure that a failed batch will be delivered. There is no end-to-end at-least-once guarantee.

Transforms

Transforms can be applied before writing to SurrealDB. See the transforms documentation for the available types and their configuration.