Sinks

Sinks receive batches from the coordinator after processors run. Each sink lives under spec.sinks and can be marked as required or best-effort via the required flag. Checkpoint behavior is governed by the pipeline’s commit policy.

Envelope and Encoding

All sinks support configurable envelope formats and wire encodings. See the Envelopes and Encodings page for detailed documentation.

Quick example:

sinks:
  - type: kafka
    config:
      id: events-kafka
      brokers: localhost:9092
      topic: events
      envelope:
        type: cloudevents
        type_prefix: "com.example.cdc"
      encoding: json

Available Sinks

	Sink	Description
	kafka	Kafka producer sink
	nats	NATS JetStream sink
	redis	Redis stream sink
	http	HTTP/Webhook sink
	s3	S3-compatible object storage (Parquet / JSON Lines) — AWS S3, MinIO, GCS, Azure, local FS
	clickhouse	ClickHouse (HTTP + RowBinary) — change-log or current-state (ReplacingMergeTree), auto table creation

Multiple sinks in one pipeline

You can combine multiple sinks in one pipeline to fan out events to different destinations. However, multi-sink pipelines introduce complexity that requires careful consideration.

Why multiple sinks are challenging

Different performance characteristics: Kafka might handle 100K events/sec while a downstream HTTP webhook processes 100/sec. The slowest sink becomes the bottleneck for the entire pipeline.

Independent failure modes: Each sink can fail independently. Redis might be healthy while Kafka experiences broker failures. Without proper handling, a single sink failure could block the entire pipeline or cause data loss.

No distributed transactions: DeltaForge cannot atomically commit across heterogeneous systems. If Kafka succeeds but Redis fails mid-batch, you face a choice: retry Redis (risking duplicates in Kafka) or skip Redis (losing data there).

Checkpoint semantics: The checkpoint represents “how far we’ve processed from the source.” With multiple sinks, when is it safe to advance? After one sink succeeds? All of them? A majority?

Read the required and commit_policy sections below for options to manage these challenges.

The required flag

The required flag on each sink determines whether that sink must acknowledge successful delivery before the checkpoint advances:

sinks:
  - type: kafka
    config:
      id: primary-kafka
      required: true    # Must succeed for checkpoint to advance
      
  - type: redis
    config:
      id: cache-redis
      required: false   # Best-effort; failures don't block checkpoint

When required: true (default): The sink must acknowledge the batch before the checkpoint can advance. If this sink fails, the pipeline blocks and retries until it succeeds or the operator intervenes.

When required: false: The sink is best-effort. Failures are logged but don’t prevent the checkpoint from advancing. Use this for non-critical destinations where some data loss is acceptable.

Commit policy

The commit_policy works with the required flag to determine checkpoint behavior:

commit_policy:
  mode: required   # Only wait for required sinks

sinks:
  - type: kafka
    config:
      required: true   # Checkpoint waits for this
  - type: redis  
    config:
      required: false  # Checkpoint doesn't wait for this
  - type: nats
    config:
      required: true   # Checkpoint waits for this

Per-sink independent checkpoints

Each sink maintains its own checkpoint, committed independently after successful delivery. This means:

• Faster sinks are not held back by slower ones while the pipeline runs - each sink advances its own checkpoint and the source’s read position does not rewind for a lagging sink in-session
• The source replays from the minimum checkpoint across all sinks. Recovery is not selective per sink: on restart every sink, including ones already ahead, is re-delivered from that minimum forward, so sinks that were ahead receive duplicates and must dedup
• Adding a new sink lowers the minimum to that sink’s earliest position, so on the next restart the source replays from there and existing sinks are re-delivered those events too (they dedup); the new sink is not backfilled in isolation
• Removing a sink cleans up its checkpoint automatically on the next pipeline patch

This architecture avoids the common CDC pitfall where the slowest sink becomes a bottleneck for all other sinks.

Delivery guarantee tiers

DeltaForge does not claim end-to-end exactly-once for any sink. Every sink is at-least-once; Kafka exactly_once: true adds transactional atomic-batch visibility on top, and NATS/Redis add a dedup mechanism. Reaching exactly-once end to end always requires consumer cooperation (dedup on the event id).

Practical patterns

Primary + secondary: One critical sink (Kafka for durability) marked required: true, with secondary sinks (Redis for caching, testing or experimentation) marked required: false.

Quorum for redundancy: Three sinks with commit_policy.mode: quorum and quorum: 2. Checkpoint advances when any two succeed, providing fault tolerance.

All-or-nothing checkpointing: Use commit_policy.mode: all when every destination is critical - the checkpoint advances only once all sinks have acknowledged (throughput is gated by the slowest sink). This gates checkpoint advancement across sinks; it is not an atomic cross-sink commit - a batch can still land in some sinks and not others within a cycle, and those are reconciled by replay on restart.

Multi-format fan-out

For sending the same events to different consumers that expect different formats:

sinks:
  # Kafka Connect expects Debezium format
  - type: kafka
    config:
      id: connect-sink
      brokers: ${KAFKA_BROKERS}
      topic: connect-events
      envelope:
        type: debezium
      required: true

  # Lambda expects CloudEvents
  - type: kafka
    config:
      id: lambda-sink
      brokers: ${KAFKA_BROKERS}
      topic: lambda-events
      envelope:
        type: cloudevents
        type_prefix: "com.acme.cdc"
      required: false

  # Analytics wants raw events
  - type: redis
    config:
      id: analytics-redis
      uri: ${REDIS_URI}
      stream: analytics
      envelope:
        type: native
      required: false

This allows each consumer to receive events in their preferred format without post-processing.