sink

type AcksLevel ¶

type AcksLevel int

AcksLevel controls how many broker acknowledgements the Kafka sink waits for before considering a write successful.

const (
	// AcksNone does not wait for any acknowledgement (fire-and-forget).
	AcksNone AcksLevel = 0
	// AcksLeader waits for the leader broker to acknowledge the write.
	AcksLeader AcksLevel = 1
	// AcksAll waits for all in-sync replicas to acknowledge the write.
	AcksAll AcksLevel = -1
)

func (a AcksLevel) Display() string

type BlackholeSink ¶

type BlackholeSink struct {
	// contains filtered or unexported fields
}

BlackholeSink discards all records. Used for benchmarking pipeline throughput without the overhead of writing to an external system.

func NewBlackholeSink() *BlackholeSink

func (s *BlackholeSink) Count() int64

func (s *BlackholeSink) Write(ctx context.Context, in <-chan types.Record) error

type CheckpointedSink ¶ added in v0.4.0

type CheckpointedSink interface {
	Sink

	// SetOnPrepared registers the coordinator callback invoked when a
	// barrier's output is fully staged (or staging failed).
	SetOnPrepared(fn func(id string, err error))

	// Commit finalizes the transaction for checkpoint id and unblocks Write.
	Commit(ctx context.Context, id string) error

	// Abort discards the transaction for checkpoint id and unblocks Write.
	Abort(ctx context.Context, id string) error

	// WasCommitted reports whether checkpoint id's transaction
	// committed before a crash.
	WasCommitted(ctx context.Context, id string) (bool, error)
}

CheckpointedSink is a Sink that participates in coordinated exactly-once checkpoints. Output between two checkpoint barriers is staged (e.g. in a Kafka transaction) and becomes visible only when the coordinator calls Commit.

Contract:

• Write sees checkpoint barriers in-band, after every pre-barrier record. On a barrier the sink must flush all staged output for the ending interval into the open transaction (plus a transaction marker identifying the checkpoint), invoke the OnPrepared callback, and BLOCK further writes until the coordinator calls Commit or Abort for that checkpoint.
• Commit makes the interval's output visible and opens the next transaction. Abort discards it.
• WasCommitted reports, after a restart, whether the transaction for a checkpoint ID actually committed (e.g. by probing the transaction marker under read-committed isolation). It resolves the prepared-but-unconfirmed recovery case.

type DLQ ¶ added in v0.4.0

type DLQ interface {
	Write(ctx context.Context, record types.Record) error
}

DLQ is a sink that receives records that could not be written to the primary sink after all retries. It is used together with FailurePolicyDLQ.

Common DLQ implementations: a separate Kafka topic, a file on disk, or a Postgres error table.

type Describable ¶

type Describable interface {
	Describe() SinkInfo
}

Describable is an optional interface that Sinks can implement to expose metadata for the dashboard.

type FailurePolicy ¶ added in v0.4.0

type FailurePolicy int

FailurePolicy determines what happens when a sink write fails after all retry attempts have been exhausted.

const (
	// FailurePolicyDrop silently discards the record.
	FailurePolicyDrop FailurePolicy = iota

	// FailurePolicyDLQ forwards the record to a dead-letter-queue sink.
	FailurePolicyDLQ

	// FailurePolicyFail returns an error and stops the pipeline.
	FailurePolicyFail
)

type JSONSerializer ¶

type JSONSerializer struct{}

JSONSerializer marshals Record.Parsed (if set) or Record.Value to JSON. When Parsed is nil and Value is already JSON bytes, this effectively passes them through (re-encoded, which is a no-op for valid JSON).

func NewJSONSerializer() *JSONSerializer

func (s *JSONSerializer) Serialize(r types.Record) ([]byte, error)

type KafkaSink ¶

type KafkaSink struct {
	// contains filtered or unexported fields
}

KafkaSink writes records to a single Kafka topic. It implements the Sink interface for use in mailer pipelines.

Configure a KafkaSink with functional options via NewKafkaSink:

sink := sink.NewKafkaSink(
    sink.KafkaSinkBrokers("localhost:9092"),
    sink.KafkaSinkTopic("order-summary"),
    sink.KafkaSinkBatchSize(200),
    sink.KafkaSinkRequiredAcks(sink.AcksAll),
    sink.KafkaSinkSASL(auth.SASLConfig{Mechanism: auth.SASLPlain, Username: "u", Password: "p"}),
    sink.KafkaSinkSerialize(sink.NewJSONSerializer()),
)

Records are written in batches for efficiency. On context cancellation, the sink drains remaining records for up to 5 seconds before flushing.

mailer.Record fields are mapped to kafka.Message as follows:

• Key -> Message.Key
• Value -> Message.Value (or serializer output if configured)
• Timestamp -> Message.Time
• Headers -> Message.Headers

func NewKafkaSink(opts ...KafkaSinkOption) *KafkaSink

func (k *KafkaSink) Describe() SinkInfo

func (k *KafkaSink) Write(ctx context.Context, in <-chan types.Record) error

type KafkaSinkOption ¶

type KafkaSinkOption func(*kafkaSinkConfig)

KafkaSinkOption configures a KafkaSink. Pass one or more to NewKafkaSink. Brokers and Topic are required.

func KafkaSinkAsync() KafkaSinkOption

func KafkaSinkBatchSize(n int) KafkaSinkOption

func KafkaSinkBatchTimeout(d time.Duration) KafkaSinkOption

func KafkaSinkBrokers(brokers ...string) KafkaSinkOption

func KafkaSinkDLQ(dlq DLQ) KafkaSinkOption

func KafkaSinkFailurePolicy(p FailurePolicy) KafkaSinkOption

func KafkaSinkMaxRetries(n int) KafkaSinkOption

func KafkaSinkRequiredAcks(level AcksLevel) KafkaSinkOption

func KafkaSinkSASL(cfg auth.SASLConfig) KafkaSinkOption

func KafkaSinkSerialize(s Serializer) KafkaSinkOption

func KafkaSinkTLS(cfg auth.TLSConfig) KafkaSinkOption

func KafkaSinkTopic(topic string) KafkaSinkOption

type PostgresSink ¶

type PostgresSink struct {
	// contains filtered or unexported fields
}

PostgresSink writes records to a Postgres database in batches. It implements the Sink interface for use in mailer pipelines.

Configure a PostgresSink with functional options via NewPostgresSink:

sink := sink.NewPostgresSink(
    sink.PostgresDSN("postgres://user:pass@localhost:5432/dbname?sslmode=disable"),
    sink.PostgresMapper(func(r types.Record) (string, []string, []any) {
        o := r.Parsed.(*Order)
        return "orders",
            []string{"order_id", "customer", "amount"},
            []any{o.OrderID, o.Customer, o.Amount}
    }),
    sink.PostgresBatchSize(500),
)

Records are accumulated and inserted in batches using multi-value INSERT statements for efficiency. On context cancellation, the sink drains remaining records for up to 5 seconds before flushing the final batch.

func NewPostgresSink(opts ...PostgresSinkOption) *PostgresSink

func (p *PostgresSink) Describe() SinkInfo

func (p *PostgresSink) Write(ctx context.Context, in <-chan types.Record) error

type PostgresSinkOption ¶

type PostgresSinkOption func(*postgresSinkConfig)

PostgresSinkOption configures a PostgresSink. Pass one or more to NewPostgresSink. DSN and a RecordMapper are required.

func PostgresBatchSize(n int) PostgresSinkOption

func PostgresConflictColumns(cols ...string) PostgresSinkOption

func PostgresDLQ(dlq DLQ) PostgresSinkOption

func PostgresDSN(dsn string) PostgresSinkOption

func PostgresFailurePolicy(p FailurePolicy) PostgresSinkOption

func PostgresFlushInterval(d time.Duration) PostgresSinkOption

func PostgresMapper(m RecordMapper) PostgresSinkOption

func PostgresMaxRetries(n int) PostgresSinkOption

func PostgresMode(mode PostgresWriteMode) PostgresSinkOption

func PostgresUpdateColumns(cols ...string) PostgresSinkOption

type PostgresWriteMode ¶ added in v0.5.0

type PostgresWriteMode string

PostgresWriteMode selects how rows are written.

const (
	PostgresInsert PostgresWriteMode = "insert"
	PostgresUpsert PostgresWriteMode = "upsert"
)

type RecordMapper ¶

type RecordMapper func(r types.Record) (table string, columns []string, values []any)

RecordMapper extracts the table name, column names, and values from a Record. The returned columns and values must be the same length and in the same order.

This is the core mapping function for the Postgres sink — it gives the user full control over how each Record maps to a database row.

Example:

mapper := func(r types.Record) (string, []string, []any) {
    o := r.Parsed.(*Order)
    return "orders",
        []string{"order_id", "customer", "amount"},
        []any{o.OrderID, o.Customer, o.Amount}
}

type Serializer ¶

type Serializer interface {
	Serialize(r types.Record) ([]byte, error)
}

Serializer converts a Record into bytes for writing to an external system. Implementations are plugged into a KafkaSink via the KafkaSinkSerialize option.

If Record.Parsed is non-nil, serializers typically serialize the typed payload. Otherwise they fall back to Record.Value (the raw bytes). The returned bytes become the Kafka message value.

type SerializerFunc ¶

type SerializerFunc func(r types.Record) ([]byte, error)

SerializerFunc lets a plain function satisfy the Serializer interface.

func (f SerializerFunc) Serialize(r types.Record) ([]byte, error)

type Sink ¶

type Sink interface {
	Write(ctx context.Context, in <-chan types.Record) error
}

Sink is where processed data leaves the pipeline. A Sink reads records from the input channel until it's closed or the context is cancelled.

type SinkInfo ¶

type SinkInfo struct {
	Type  string            `json:"type"`
	Props map[string]string `json:"props"`
}

SinkInfo holds display metadata about a sink.

type StdoutSink ¶

type StdoutSink struct{}

StdoutSink prints each record to stdout in a human-readable format. Useful for debugging and examples.

func NewStdoutSink() *StdoutSink

func (s *StdoutSink) Describe() SinkInfo

func (s *StdoutSink) Write(ctx context.Context, in <-chan types.Record) error

type TxnKafkaOption ¶ added in v0.4.0

type TxnKafkaOption func(*txnKafkaConfig)

TxnKafkaOption configures a TxnKafkaSink.

func TxnKafkaBrokers(brokers ...string) TxnKafkaOption

func TxnKafkaMarkerTopic(topic string) TxnKafkaOption

func TxnKafkaSASL(cfg auth.SASLConfig) TxnKafkaOption

func TxnKafkaSerialize(s Serializer) TxnKafkaOption

func TxnKafkaTLS(cfg auth.TLSConfig) TxnKafkaOption

func TxnKafkaTopic(topic string) TxnKafkaOption

func TxnKafkaTransactionalID(id string) TxnKafkaOption

type TxnKafkaSink ¶ added in v0.4.0

type TxnKafkaSink struct {
	// contains filtered or unexported fields
}

TxnKafkaSink writes records to Kafka inside transactions, one transaction per checkpoint interval, implementing CheckpointedSink for end-to-end exactly-once pipelines:

source offsets + operator state + sink output commit atomically.

Built on franz-go (segmentio/kafka-go has no transactional producer). Every checkpoint transaction also carries a marker record in the marker topic (default "<topic>.checkpoints"); after a crash, the marker's visibility under read_committed proves whether the transaction committed — this resolves prepared-but-unconfirmed checkpoints without Kafka transaction resumption.

IMPORTANT: downstream consumers of the output topic MUST use isolation.level=read_committed, or they will observe records from aborted transactions. Records become visible only when the checkpoint interval's transaction commits — the checkpoint interval is therefore also the output visibility latency.

Usage:

sk := sink.NewTxnKafkaSink(
    sink.TxnKafkaBrokers("localhost:9092"),
    sink.TxnKafkaTopic("order-summary"),
    sink.TxnKafkaTransactionalID("order-pipeline-1"),
)

The transactional ID must be unique per pipeline instance; a second instance with the same ID fences the first (Kafka zombie fencing).

func NewTxnKafkaSink(opts ...TxnKafkaOption) *TxnKafkaSink

func (s *TxnKafkaSink) Abort(ctx context.Context, id string) error

func (s *TxnKafkaSink) Commit(ctx context.Context, id string) error

func (s *TxnKafkaSink) Describe() SinkInfo

func (s *TxnKafkaSink) SetOnPrepared(fn func(id string, err error))

func (s *TxnKafkaSink) TransactionalID() string

func (s *TxnKafkaSink) WasCommitted(ctx context.Context, id string) (bool, error)

func (s *TxnKafkaSink) Write(ctx context.Context, in <-chan types.Record) error