crabka-connect

Connector-framework SPI for Crabka sources, sinks, converters, and embedded runtimes.

Part of Crabka, a Rust implementation of Apache Kafka-compatible infrastructure and clients.

Overview

crabka-connect defines the traits and data model for connector authors. It is the shared SPI for CDC sources, telemetry sinks, byte converters, typed schema converters, connector configuration, and the single-process runtime that pipes a source into a sink.

The crate is intentionally embeddable: there is no Kafka Connect worker protocol and no REST management server. Build connectors programmatically and run them inside the process that owns their lifecycle.

Capabilities

Source<K, V> for polling records from an external system, checkpointing read position, seeking on restart, and acknowledging committed checkpoints.
Sink<K, V> for writing batches, flushing, and optionally bracketing writes in transactions.
ConnectRecord<K, V> with optional key/value, timestamp, headers, and source offset metadata.
Converter<T> for crossing between typed connector payloads and Kafka wire bytes.
ByteIdentity for byte-for-byte passthrough.
SchemaConverter<T> for Confluent-framed Avro, Protobuf, or JSON payloads via crabka-schema-serde.
ConnectorRuntime for the sequential poll -> put -> commit -> checkpoint -> acknowledge loop.
ConfigDef, ConnectorConfig, secret resolution, typed config extraction, and a default derive macro.

Runtime Model

ConnectorRuntime owns one source and one sink. Each interval polls a bounded batch, writes it to the sink, commits or flushes the sink, persists the source checkpoint, then acknowledges the source only after the checkpoint is durable.

This keeps the source from running ahead of records the sink has committed. The included InMemoryCheckpointStore is useful for tests and short-lived tools; production connectors should provide a durable CheckpointStore.

Install

cargo add crabka-connect
cargo add serde_json

For workspace development, use the path dependency from this repository.

Usage

Define typed connector configuration with the default derive feature:

use crabka_connect::{ConfigDef, ConnectorConfig, EnvSecretResolver, SecretString};
use serde_json::json;

#[derive(ConnectorConfig)]
struct PostgresSourceConfig {
    #[config(required)]
    database_url: String,

    #[config(secret)]
    password: SecretString,

    #[config(default = "public")]
    schema: String,
}

# async fn build() -> crabka_connect::ConfigResult<PostgresSourceConfig> {
let raw = serde_json::Map::from_iter([
    ("database_url".to_string(), json!("postgres://localhost/app")),
    (
        "password".to_string(),
        json!({ "from": "env", "name": "POSTGRES_PASSWORD" }),
    ),
]);

let def: ConfigDef = PostgresSourceConfig::config_def();
let resolved = def.resolve(raw, &EnvSecretResolver).await?;
let config = PostgresSourceConfig::from_resolved(&resolved)?;
# Ok(config)
# }

Secret fields resolve through SecretResolver implementations and are redacted by Debug and Display. Literal secrets are rejected unless explicitly allowed with ResolveOptions.

Cargo Features

derive - enables and reexports crabka-connect-derive::ConnectorConfig. This feature is enabled by default.

Documentation

License

Apache-2.0. Derivative work of Apache Kafka; see NOTICE.

crabka-connect 0.3.9