source

package
v0.7.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 19, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func KafkaToRecord 

func KafkaToRecord(msg kafka.Message) types.Record

KafkaToRecord converts a kafka.Message to a mailer.Record. Headers are copied into a map. Parsed is left nil here; the deserializer (if configured) populates it in runOnce after this conversion.

Types

type CheckpointSource 

type CheckpointSource interface {
	// CheckpointOffset returns the source's current position as opaque bytes.
	CheckpointOffset() ([]byte, error)

	// RestoreOffset seeks the source to the position saved by CheckpointOffset.
	RestoreOffset(data []byte) error
}

CheckpointSource is an optional interface that Sources can implement to support checkpointing. When the CheckpointCoordinator needs to create a checkpoint, it asks the source to save its current offset so it can resume from that point on recovery.

type CommitPolicy added in v0.4.0 

type CommitPolicy int

CommitPolicy determines what happens when a Kafka offset commit fails after all retry attempts have been exhausted. 

const (
	// CommitPolicySkip logs the error and continues processing.
	// Records may be reprocessed on restart because the offset was
	// never committed.
	CommitPolicySkip CommitPolicy = iota

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

type Describable 

type Describable interface {
	Describe() SourceInfo
}

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

type DeserFailurePolicy added in v0.4.0 

type DeserFailurePolicy int

DeserFailurePolicy determines what happens when deserialization fails for a record read from the source. 

const (
	// DeserFailureDrop silently drops the record.
	DeserFailureDrop DeserFailurePolicy = iota

	// DeserFailureDLQ forwards the raw record to a dead-letter queue.
	DeserFailureDLQ

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

type Deserializer 

type Deserializer interface {
	Deserialize(data []byte, headers map[string][]byte) (any, error)
}

Deserializer converts raw Kafka bytes into a typed payload. Implementations are plugged into a KafkaSource via the KafkaDeserialize option.

If Deserialize returns an error, the record is dropped from the stream. The returned value is stored in Record.Parsed; Record.Value keeps the raw bytes.

type DeserializerFunc 

type DeserializerFunc func(data []byte, headers map[string][]byte) (any, error)

DeserializerFunc lets a plain function satisfy the Deserializer interface.

func (DeserializerFunc) Deserialize 

func (f DeserializerFunc) Deserialize(data []byte, headers map[string][]byte) (any, error)

type Drainable added in v0.4.0 

type Drainable interface {
	Drain(ctx context.Context) error
}

Drainable is an optional interface that sources implement to flush pending state (e.g. uncommitted Kafka offsets) during graceful shutdown.

type GeneratorSource 

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

GeneratorSource produces a fixed slice of records and then closes. Useful for testing and examples.

func FromSlices 

func FromSlices(keys []string, values []string) *GeneratorSource

FromSlices is a convenience function that creates records from string key-value pairs. Each record gets an incrementing offset and the current timestamp.

func NewGeneratorSource 

func NewGeneratorSource(records []types.Record) *GeneratorSource

NewGeneratorSource creates a source that emits the given records in order.

func (*GeneratorSource) Run 

func (s *GeneratorSource) Run(ctx context.Context, out chan<- types.Record) error

Run emits all records into the output channel and then returns. The channel owner (StreamExecutionEnv) is responsible for closing it.

type JSONDeserializer 

type JSONDeserializer[T any] struct{}

JSONDeserializer unmarshals each message's Value into a fresh T and returns a *T as the Parsed payload.

Usage: 

deser := source.NewJSONDeserializer[Order]()
src := source.NewKafkaSource(
    source.KafkaBrokers("localhost:9092"),
    source.KafkaTopic("orders"),
    source.KafkaDeserialize(deser),
)

func NewJSONDeserializer 

func NewJSONDeserializer[T any]() *JSONDeserializer[T]

NewJSONDeserializer creates a JSON deserializer for type T.

func (*JSONDeserializer[T]) Deserialize 

func (d *JSONDeserializer[T]) Deserialize(data []byte, _ map[string][]byte) (any, error)

type KafkaSource 

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

KafkaSource reads records from one or more Kafka topics using a consumer group. It implements the Source interface for use in mailer pipelines, and the CheckpointSource interface for checkpoint/recovery support.

Configure a KafkaSource with functional options via NewKafkaSource: 

src := source.NewKafkaSource(
    source.KafkaBrokers("localhost:9092"),
    source.KafkaTopic("orders"),
    source.KafkaGroupID("order-processor"),
    source.KafkaStartFrom(source.OffsetEarliest),
    source.KafkaParallel(),
    source.KafkaWithWatermarks(1*time.Second),
)

Kafka message fields are mapped to mailer.Record as follows:

  • Key -> Record.Key
  • Value -> Record.Value
  • Time -> Record.Timestamp
  • Offset -> Record.Offset
  • Headers -> Record.Headers

If a Deserializer is configured (KafkaDeserialize), Record.Parsed is also set.

func NewKafkaSource 

func NewKafkaSource(opts ...KafkaSourceOption) *KafkaSource

NewKafkaSource creates a Source that reads from Kafka. Brokers and at least one topic (via KafkaTopic or KafkaTopics) are required; if missing, NewKafkaSource panics to fail fast at construction time.

func (*KafkaSource) CheckpointOffset 

func (k *KafkaSource) CheckpointOffset() ([]byte, error)

CheckpointOffset returns the source's current position as JSON bytes. For Kafka, this is the last committed offset per partition (from reader stats).

func (*KafkaSource) CommitOffsets added in v0.4.0 

func (k *KafkaSource) CommitOffsets(ctx context.Context, data []byte) error

CommitOffsets implements source.OffsetCommitter: commits the given barrier-aligned offsets to the broker after a coordinated checkpoint completes. Advisory (consumer-lag visibility) — recovery reads the checkpoint file, never the broker. Only possible in consumer-group mode; parallel per-partition readers have no group to commit to.

func (*KafkaSource) Describe 

func (k *KafkaSource) Describe() SourceInfo

Describe returns metadata about this Kafka source for the dashboard.

func (*KafkaSource) Drain added in v0.4.0 

func (k *KafkaSource) Drain(ctx context.Context) error

Drain flushes pending offset commits. Called during graceful shutdown to commit offsets for records that were read but not yet committed.

func (*KafkaSource) RestoreOffset 

func (k *KafkaSource) RestoreOffset(data []byte) error

RestoreOffset restores per-partition offsets from a checkpoint.

func (*KafkaSource) Run 

func (k *KafkaSource) Run(ctx context.Context, out chan<- types.Record) error

Run fetches messages from Kafka and writes them to the output channel until the context is cancelled or the reader returns an error.

If KafkaWithWatermarks was set, Run applies watermarks by delegating to a WatermarkSource wrapping this source — callers always get a watermark-aware stream when the option is present.

type KafkaSourceOption 

type KafkaSourceOption func(*kafkaSourceConfig)

KafkaSourceOption configures a KafkaSource. Pass one or more to NewKafkaSource. Brokers and at least one topic are required.

func KafkaBrokers 

func KafkaBrokers(brokers ...string) KafkaSourceOption

KafkaBrokers sets the Kafka bootstrap brokers. Required. Example: KafkaBrokers("localhost:9092") or KafkaBrokers("a:9092", "b:9092").

func KafkaCommitBatch 

func KafkaCommitBatch(n int) KafkaSourceOption

KafkaCommitBatch batches offset commits every N messages instead of committing after every single message. Pass 0 (default) to commit per-message. Higher values improve throughput but risk reprocessing on failure within the batch window.

func KafkaCommitMaxRetries added in v0.4.0 

func KafkaCommitMaxRetries(n int) KafkaSourceOption

KafkaCommitMaxRetries sets how many times CommitMessages is retried before the commit failure policy is applied (default 3).

func KafkaCommitPolicy added in v0.4.0 

func KafkaCommitPolicy(p CommitPolicy) KafkaSourceOption

KafkaCommitPolicy sets what happens when offset commits fail after all retries. Default is CommitPolicySkip (log and continue).

func KafkaDeserialize 

func KafkaDeserialize(d Deserializer) KafkaSourceOption

KafkaDeserialize sets a Deserializer that runs on every record before it is emitted. The deserialized value is stored in Record.Parsed. If the deserializer returns an error, the record is dropped by default. Use KafkaDeserializeFailurePolicy to control this behavior.

func KafkaDeserializeDLQ added in v0.4.0 

func KafkaDeserializeDLQ(dlq RecordSink) KafkaSourceOption

KafkaDeserializeDLQ sets the dead-letter-queue for records that fail deserialization. Only used with DeserFailureDLQ.

func KafkaDeserializeFailurePolicy added in v0.4.0 

func KafkaDeserializeFailurePolicy(p DeserFailurePolicy) KafkaSourceOption

KafkaDeserializeFailurePolicy sets what happens when deserialization fails. Default is DeserFailureDrop (record is silently discarded).

func KafkaExactlyOnce added in v0.4.0 

func KafkaExactlyOnce() KafkaSourceOption

KafkaExactlyOnce puts the source in exactly-once mode for use with a CheckpointedSink (e.g. TxnKafkaSink):

  • Eager broker offset commits are DISABLED. Offsets are committed to the broker only after a coordinated checkpoint completes, and even then only for consumer-lag visibility — recovery reads offsets from the checkpoint, not the broker.
  • Reads use isolation.level=read_committed, so upstream aborted transactions are never consumed.

func KafkaFetchBytes 

func KafkaFetchBytes(min, max int) KafkaSourceOption

KafkaFetchBytes controls the min/max bytes returned in each fetch request. Defaults: min=1, max=10MB. Increase max for higher throughput on large messages.

func KafkaFetchMaxRetries added in v0.4.0 

func KafkaFetchMaxRetries(n int) KafkaSourceOption

KafkaFetchMaxRetries sets how many times FetchMessage is retried before failing the pipeline (default 5). Pass -1 to retry forever.

func KafkaGroupID 

func KafkaGroupID(id string) KafkaSourceOption

KafkaGroupID sets the consumer group ID. When set, offsets are committed to Kafka and tracked by the broker. When empty, no commits happen.

func KafkaParallel added in v0.4.0 

func KafkaParallel() KafkaSourceOption

KafkaParallel enables per-partition parallel reading. Each partition gets its own goroutine and kafka.Reader, so records from different partitions are fetched concurrently. This improves throughput for multi-partition topics.

Note: enabling parallel disables consumer-group auto-balancing. Offsets are managed per partition.

func KafkaSASL 

func KafkaSASL(cfg auth.SASLConfig) KafkaSourceOption

KafkaSASL enables SASL authentication on the Kafka connection.

func KafkaStartFrom 

func KafkaStartFrom(o Offset) KafkaSourceOption

KafkaStartFrom sets where consumption begins. Use OffsetEarliest or OffsetLatest. Defaults to OffsetEarliest.

func KafkaTLS 

func KafkaTLS(cfg auth.TLSConfig) KafkaSourceOption

KafkaTLS enables TLS on the Kafka connection.

func KafkaTopic 

func KafkaTopic(topic string) KafkaSourceOption

KafkaTopic sets a single topic to consume from. Either KafkaTopic or KafkaTopics is required.

func KafkaTopics 

func KafkaTopics(topics ...string) KafkaSourceOption

KafkaTopics sets multiple topics to consume from (consumer-group mode). Takes precedence over KafkaTopic.

func KafkaWatermarkInterval 

func KafkaWatermarkInterval(d time.Duration) KafkaSourceOption

KafkaWatermarkInterval overrides the default 500ms watermark emission interval. Only meaningful when KafkaWithWatermarks is also set.

func KafkaWithWatermarks 

func KafkaWithWatermarks(maxOutOfOrderness time.Duration) KafkaSourceOption

KafkaWithWatermarks enables automatic watermark injection so that downstream windows fire without manually wrapping the source. maxOutOfOrderness is the maximum expected delay between events; the watermark is set to (maxTimestampSeen - maxOutOfOrderness).

type Offset 

type Offset int64

Offset is a mailer-agnostic Kafka offset position. It replaces direct use of kafka-go's offset constants so users never need to import segmentio/kafka-go. 

const (
	// OffsetEarliest starts reading from the earliest available message
	// in each partition (maps to kafka.FirstOffset).
	OffsetEarliest Offset = -2

	// OffsetLatest starts reading from the most recently produced message
	// in each partition (maps to kafka.LastOffset).
	OffsetLatest Offset = -1
)

type OffsetCommitter added in v0.4.0 

type OffsetCommitter interface {
	CommitOffsets(ctx context.Context, offsets []byte) error
}

OffsetCommitter is an optional interface for sources whose offsets should be committed externally (e.g. to the Kafka broker) after a coordinated checkpoint completes. The commit is ADVISORY — consumer lag visibility only. The checkpoint file remains the source of truth for recovery. Offsets use the same JSON shape as CheckpointSource.CheckpointOffset ({"partition": nextOffset}).

type OffsetSpec 

type OffsetSpec struct {
	Mode     Offset
	Explicit map[int]int64 // partition -> offset (only used when Mode == offsetExplicit)
}

OffsetSpec describes where a Kafka source should start consuming. It is set via the KafkaStartFrom option.

func (OffsetSpec) Display added in v0.3.0 

func (s OffsetSpec) Display() string

Display returns a human-readable offset description for the dashboard.

type RecordSink added in v0.4.0 

type RecordSink interface {
	Write(ctx context.Context, r types.Record) error
}

RecordSink receives failed records for dead-letter-queue routing. Implemented by any sink (KafkaSink, PostgresSink, etc.).

type SliceSource 

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

SliceSource reads records from an in-memory slice. Unlike GeneratorSource, SliceSource accepts a raw []Record slice directly — no need to call NewGeneratorSource or FromSlices first.

It emits each record in order and then closes. Useful for tests and one-shot pipelines where you already have the data in memory.

func NewSliceSource 

func NewSliceSource(records []types.Record) *SliceSource

NewSliceSource creates a source that emits the given records in order.

func (*SliceSource) Describe 

func (s *SliceSource) Describe() SourceInfo

Describe returns metadata for the dashboard.

func (*SliceSource) Run 

func (s *SliceSource) Run(ctx context.Context, out chan<- types.Record) error

Run emits all records into the output channel and then returns. The channel owner (StreamExecutionEnv) is responsible for closing it.

type Source 

type Source interface {
	Run(ctx context.Context, out chan<- types.Record) error
}

Source is where data enters the pipeline. A Source continuously emits Records into the output channel until the context is cancelled or the source is exhausted. The channel owner (e.g. StreamExecutionEnv) is responsible for closing the output channel.

type SourceInfo 

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

SourceInfo holds display metadata about a source.

type WatermarkSource 

type WatermarkSource struct {
	Source    Source
	Generator watermark.WatermarkGenerator
	Interval  time.Duration
}

WatermarkSource wraps a Source and injects watermark records into the stream based on a WatermarkGenerator. The source tracks the maximum event timestamp seen and periodically emits watermarks to drive window completion.

func NewWatermarkSource 

func NewWatermarkSource(src Source, gen watermark.WatermarkGenerator, interval time.Duration) *WatermarkSource

NewWatermarkSource wraps a Source with watermark generation. Every interval, it checks if the generator has a new watermark and injects it.

func (*WatermarkSource) Run 

func (ws *WatermarkSource) Run(ctx context.Context, out chan<- types.Record) error

Run starts the underlying source, intercepts every record to update the watermark generator, and periodically injects watermark records. The channel owner is responsible for closing the output channel.

On context cancellation or source completion, a max watermark is emitted to flush all remaining windows. This guarantees correct end-of-stream behavior for both batch and streaming sources.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.