Custom sources

A source is two pieces, split on purpose:

• Source — the control plane. Lifecycle, lane assignment/revocation events, watermark commits, pause/resume. Driven by the runtime's controller from a single thread.
• SourceLane — the data plane. A pollable unit pinned to one pipeline thread, yielding payload batches that borrow the lane's buffers.

The API is poll-based rather than a futures::Stream, deliberately: real source clients (librdkafka) already own their network I/O on their own threads, and a Stream wrapper only adds waker overhead plus 'static bounds that break payload borrowing. Rationale in docs/DESIGN.md (§ Source abstraction); trait docs in crates/etl-core/src/source/mod.rs and on docs.rs.

This page mirrors crates/etl/examples/custom_source_sink.rs — a generator source counting to a limit per partition. Run it:

cargo run -p etl --example custom_source_sink

The data plane: SourceLane and PayloadBatch

A lane polls up to max_records payloads, waiting at most timeout:

impl SourceLane for CounterLane {
    type Batch<'a> = CounterBatch<'a>;

    fn id(&self) -> LaneId { self.id }
    fn partition(&self) -> PartitionId { self.partition }

    fn poll(
        &mut self,
        max_records: usize,
        timeout: Duration,
    ) -> Result<Option<Self::Batch<'_>>, SourceError> {
        if self.next >= self.limit {
            // Exhausted: block briefly like an idle consumer would — a lane
            // must never busy-spin the pipeline thread.
            std::thread::sleep(timeout);
            return Ok(None);
        }
        let base = self.next;
        let end = (base + max_records as i64).min(self.limit);
        self.buf.clear();
        self.buf.extend((base..end).map(|n| n.to_string().into_bytes()));
        self.next = end;

        // One acknowledgement handle per batch.
        let ack = self.issuer.issue(self.partition, end - 1);
        Ok(Some(CounterBatch {
            payloads: &self.buf,
            partition: self.partition,
            base_offset: base,
            idx: 0,
            ack,
        }))
    }
}

What the framework holds you to:

• Payloads borrow the lane. Batch is a GAT borrowing '_ from the lane, so payloads can point straight into your client's buffers — zero copies out of the source. The batch is dropped before the next poll on the same lane; records derived from it are consumed or encoded within that window (the operator chain guarantees this by construction).
• One AckRef per batch. Issue it through the AckIssuer you received at open (issuer.issue(partition, last_offset) — the offset of the last record in the batch). Every record deserialized from the batch clones this handle; the checkpointer advances the partition's watermark past the batch only when every clone resolves as delivered. You never track individual records.
• Idle lanes block, never spin. Ok(None) means nothing arrived — but reach it by waiting up to timeout (a blocking queue read, a condvar, worst case a sleep). A zero-timeout poll loop busy-spins a pinned pipeline thread at 100% for nothing.

The batch type streams payloads out one at a time, all sharing the batch lifetime:

impl<'a> PayloadBatch<'a> for CounterBatch<'a> {
    fn next_payload(&mut self) -> Option<RawPayload<'a>> {
        let bytes = self.payloads.get(self.idx)?;
        let offset = self.base_offset + self.idx as i64;
        self.idx += 1;
        Some(RawPayload { bytes, key: None, partition: self.partition,
                          offset, timestamp_ms: offset })
    }

    fn ack(&self) -> &AckRef { &self.ack }
}

RawPayload.key feeds the sink's shard routing hash — populate it if your source has message keys.

The control plane: Source

impl Source for CounterSource {
    type Lane = CounterLane;

    fn open(&mut self, ctx: SourceCtx) -> Result<(), SourceError> {
        self.issuer = Some(ctx.issuer);   // clone into every lane you build
        Ok(())
    }

    fn poll_events(&mut self, timeout: Duration) -> Result<SourceEvent<CounterLane>, SourceError> {
        if !self.handed_out {
            self.handed_out = true;
            let lanes = /* build one CounterLane per partition */;
            return Ok(SourceEvent::LanesAssigned(lanes));
        }
        std::thread::sleep(timeout);      // nothing else ever happens
        Ok(SourceEvent::Idle)
    }

    fn commit(&mut self, watermarks: &[(PartitionId, i64)]) -> Result<(), SourceError> {
        // Store durably per your source's policy (Kafka: store_offsets).
        Ok(())
    }
}

• open runs once, before anything else, and hands you the AckIssuer. Connect here.
• poll_events services control-plane work (rebalance callbacks, statistics) and returns the next SourceEvent: LanesAssigned(lanes), LanesRevoked { lanes, barrier }, or Idle. The runtime calls it regularly regardless of backpressure state — pausing data never pauses the control plane, which is how group liveness survives a slow sink.
• commit receives per-partition committable watermarks. Each is one past the last acknowledged offset — store it as-is (this matches Kafka's commit convention). Interval-driven; durability is per your source's own policy.
• flush_commits (default no-op) must synchronously flush stored positions — the runtime calls it at shutdown and revocation.
• pause/resume (default no-ops) are the backpressure hooks: stop and restart fetching for specific lanes. Sources that can't pause rely on bounded-queue pushback alone, but a pausable source keeps memory flat under sink pressure.

Epochs and rebalances

Assignment changes are where at-least-once is won or lost:

• LanesAssigned bumps the assignment epoch in the checkpointer, and acks issued under a previous epoch are discarded — a stale ack from a revoked assignment can never advance a new epoch's watermark.
• Assignment is eager: begin_epoch replaces all trackers, so a LanesAssigned event must describe the full assignment following a full revocation, not a delta. (Incremental/cooperative assignment is future work; the controller defensively drains and commits live lanes if a source violates this.)
• LanesRevoked carries a DrainBarrier. The runtime trips it for the owning pipeline threads, which stop the lanes, flush in-flight records, and arrive at the barrier. Complete the revocation — final synchronous commit included — only after DrainBarrier::wait returns. This is the choreography that lets a scale-down cost only the uncommitted tail; see Delivery guarantees.

[!WARNING] Never block inside poll_events waiting on data-plane progress you yourself gate. The one sanctioned wait is DrainBarrier::wait during a revocation — the runtime guarantees the pipeline threads are draining toward it.

Testing

Drive your source without a full pipeline: construct a Checkpointer, call open with SourceCtx::new(cp.handle()), poll events, poll lanes, and assert on committed watermarks — the bottom half of crates/etl/examples/custom_operator.rs does exactly this with the memory source, and Testing pipelines covers the etl-test harness. To run it for real, hand it to the builder: pipeline.sink(...)?.chains(...).run(source) — the source stays a terminal generic, so no registration is needed (Assembling a pipeline).