Skip to main content

Custom sinks

A sink is two pieces, split across the process's two halves (Architecture):

  • RowEncoder — the CPU half. Turns one record into wire bytes. Runs on pipeline threads, where payload borrows are still valid.
  • ShardWriter — the I/O half. Writes one sealed batch to one replica endpoint. Runs on sink workers in the I/O runtime.

Everything between is the framework's: chunking, per-shard batching to max_rows/max_bytes/linger, replica rotation, circuit breakers, retries, acknowledgements, backpressure. You write the two edges.

This page mirrors the sink half of crates/etl/examples/custom_source_sink.rs (a JSON-lines-to-stdout sink) and the bundle seam in crates/etl-core/src/sink/bundle.rs.

The CPU half: RowEncoder​

#[derive(Clone)]
struct JsonLinesEncoder;

impl RowEncoder<Owned<Vec<u8>>> for JsonLinesEncoder {
    fn encode<'buf>(
        &mut self,
        rec: &Record<Vec<u8>>,
        buf: &mut bytes::BytesMut,
    ) -> Result<(), SinkError> {
        use bytes::BufMut;
        buf.put_slice(b"{\"partition\":");
        buf.put_slice(rec.meta.partition.0.to_string().as_bytes());
        buf.put_slice(b",\"value\":");
        buf.put_slice(&rec.payload);
        buf.put_slice(b"}\n");
        Ok(())
    }
}

Contracts:

  • No I/O, ever. encode runs inside the hot loop on a pinned pipeline thread. A network call or blocking syscall here stalls polling for every lane on that thread.
  • Frames must concatenate. Chunks encoded on different pipeline threads are merged by shard workers into one batch, so your wire format must be valid as a plain concatenation of row encodings (RowBinary and NDJSON are; a format needing a global header belongs in the writer).
  • Errors are record-level by default, subject to the sink stage's ErrorPolicy (ChunkConfig::encode_policy, default Skip — counted in metrics). Return an error classed Fatal to stop the pipeline regardless of policy: fatal means the encoder is broken (e.g. the row type can never match the target schema) and every subsequent record would fail identically. See Error handling.

The I/O half: ShardWriter​

struct StdoutWriter;

impl ShardWriter for StdoutWriter {
    type Endpoint = String; // a real sink holds a connected client here

    fn write_batch(
        &self,
        endpoint: &String,
        batch: &SealedBatch,
    ) -> impl Future<Output = Result<(), SinkError>> + Send {
        async move {
            /* write batch.frames to the endpoint */
            Ok(())
        }
    }
}
  • Ok is the durable-ack point. Return Ok(()) only after the target system has durably accepted the batch (for ClickHouse, the successful server ack; here, after printing). The framework releases the batch's acknowledgements on Ok, which is what lets the source watermark advance past that data — return Ok early and a crash loses records, violating at-least-once.
  • Retries reuse the sealed batch unchanged, rotating across healthy replicas with capped exponential backoff. SealedBatch::dedup_token is deterministic per batch: pass it to the target if it supports deduplication so same-boundary retries are idempotent (see Delivery guarantees for what dedup tokens do and do not cover).
  • Classify errors honestly. Retryable errors get retried; a Fatal error abandons the batch (its acks fail, the partition watermark stalls, and checkpoint.stalled_fail_after eventually fails the pipeline so the data replays after restart).
  • One writer instance is shared by every shard worker (Send + Sync); the per-replica state lives in Endpoint.

Handing it to the builder: SinkParts​

Hand-rolled sinks need no trait impl — SinkParts is the bundle:

let sink = SinkParts::new(
    StdoutWriter,
    // [shard][replica]: two shards, one replica each.
    vec![vec!["shard-0".to_string()], vec!["shard-1".to_string()]],
    SinkPoolConfig::default(),
)
.with_component_type("stdout");

let runtime = pipeline.sink(sink)?/* .chains(...) ... */;
  • SinkParts::new(writer, shard_endpoints, pool) — the endpoints are indexed [shard][replica] and must be non-empty with at least one replica per shard; the builder rejects ragged or empty topologies before anything spawns. SinkPoolConfig carries the batching, in-flight, retry, and breaker tuning (Tuning).
  • .with_component_type("...") — sets the component_type metric label (default "custom") on every sink series (Monitoring).
  • .with_replica_labels(...) — display labels for per-replica metrics, same [shard][replica] shape; defaults to "{component_type}-{shard}-{replica}".
  • .with_probe(...) — an optional repeatable connectivity probe. The runtime polls it at startup and periodically, driving the sinks-connected half of /readyz — without one, readiness can't reflect your sink. Probes should use their own client set: sharing the writer's connections would report the insert path healthy simply because probing keeps it warm.

The struct is #[non_exhaustive] — construct via new plus the with_* methods, so fields can grow without breaking you. Queue depth between pipeline threads and shard workers is the builder's knob, not the bundle's: Pipeline::sink_with(bundle, SinkOptions::default().with_queue_capacity(n)).

The connector pattern: implementing SinkBundle​

A reusable connector (the etl-clickhouse pattern) exposes a config-built sink type and implements SinkBundle on it, so applications write one line:

impl SinkBundle for MySink {
    type Writer = MyShardWriter;

    fn into_parts(self) -> SinkParts<MyShardWriter> {
        SinkParts::new(self.writer, self.shard_endpoints, self.pool)
            .with_component_type("mydb")
            .with_probe(self.probe)
    }
}

Pipeline::sink accepts anything implementing SinkBundle — SinkParts itself implements it, which is why hand-rolled sinks skip the impl. The only bound anywhere is ShardWriter, so your connector's client types never appear in framework APIs (the dependency policy in docs/DESIGN.md). Connector-specific flows the framework cannot name — like ClickHouse's validate_schema producing the encoder's row schema — stay concrete pre-steps in your factory, before into_parts.

Testing​

etl-test drives chains end to end against capturing sinks, and the assertions in custom_source_sink.rs (observe commits, then trigger shutdown) are the template for verifying your durable-ack point actually gates the watermark — see Testing pipelines.