mailer

package module
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Jul 9, 2026 License: MIT Imports: 10 Imported by: 0

README

Mailer

A stream processing engine in Go, inspired by Apache Flink.

Mailer reads unbounded data streams from sources like Apache Kafka, applies real-time transformations (map, filter, reduce, window, join), and writes results to sinks. It supports stateful processing, windowed aggregations, and fault tolerance via checkpointing.

Why?

Apache Flink is powerful but Java-heavy and complex. There's no idiomatic Go stream processing engine that lets you:

  • Consume Kafka topics as unbounded streams
  • Apply stateful transformations with exactly-once semantics
  • Window and aggregate events by time
  • Recover from failures without data loss

Mailer fills this gap. It's a lightweight, embeddable Go library — not a cluster runtime. You import it, define your pipeline, and run it.

Core Concepts

Stream

An unbounded, ordered sequence of records. A stream is created from a Source and flows through a chain of Operators.

Source[Kafka] → Map → Filter → Window(5min) → Reduce → Sink[Postgres]
Record

A single data item flowing through a stream. Every record carries:

Field Type Description
Key []byte Partition key (used for keyed state and shuffling)
Value []byte The actual data payload
Timestamp time.Time Event timestamp (for time-based operations)
Offset int64 Source offset (for checkpointing and replay)
Headers map[string][]byte Optional metadata headers
Source

Where data enters the pipeline. A Source reads from an external system and emits Records into the stream.

Source Description
KafkaSource Consumes from one or more Kafka topics with consumer group support
GeneratorSource Generates synthetic records for testing
SliceSource Reads from an in-memory slice (for testing)
Sink

Where results leave the pipeline. A Sink receives processed Records and writes them to an external system.

Sink Description
KafkaSink Produces to a Kafka topic (SASL/TLS, batch writes, serializer)
PostgresSink Batch inserts into Postgres tables (pgx, retry, full mapper)
StdoutSink Prints records to stdout (debugging)
BlackholeSink Discards everything (benchmarking)
Operator

A transformation applied to a stream. Each operator takes one or more input streams and produces one or more output streams.

Operator Signature Description
Map func(Record) Record Transform each record 1:1
FlatMap func(Record) []Record Transform each record 1:many
Filter func(Record) bool Keep or drop records
KeyBy func(Record) []byte Partition stream by key (enables keyed state)
Reduce func(accum, current Record) Record Combine records with shared state
Window WindowAssigner Group records into time-based windows
Process ProcessFunction Full control: access state, timers, side outputs
Keyed Stream

After KeyBy, the stream is partitioned by key. Each key gets its own state — this is how you do per-user counters, per-session windows, etc. State is always local to the key, no cross-key coordination needed.

stream.KeyBy(func(r Record) []byte { return r.Key })
       .Reduce(func(accum, curr Record) Record { /* merge */ })
Window

Windows group an unbounded stream into finite chunks based on time. Without windows, aggregations like "count" or "sum" would never complete — the stream never ends.

Window Type Behavior Example
Tumbling Fixed-size, non-overlapping 5-minute tumbling: [0-5), [5-10), [10-15)
Sliding Fixed-size, overlapping with slide 5-min window sliding every 1 min
Session Gap-based, variable size Inactivity gap of 30 seconds

When a window closes, all records in that window are passed to the window function (Reduce, Process, etc.).

Watermark

A watermark is a timestamp that says "no records with timestamp < X will arrive after this point." Watermarks are how Mailer decides when a window is complete. If a record arrives after the watermark has passed its timestamp, it's late — and can be dropped or handled separately.

Records:    e1(2)  e2(5)  e3(8)  ---watermark(6)---  e4(7)  e5(10)
                                            ↑
                                    e3 is on-time
                                    e4 is late (7 < watermark 6)
                                    e5 is on-time

Watermarks are generated by the Source. For Kafka, a simple strategy is: watermark = max_timestamp_seen - allowed_lateness.

State

State is memory that persists across records. It's what makes a stream processor stateful.

State Type Use Case Example
ValueState Single value per key Current user score
ListState Ordered list per key Recent login timestamps
MapState Key-value map per key Per-URL click counts for a user

State is stored in a State Backend (in-memory for v1, pluggable later). On checkpoint, state is snapshotted so it can be restored after a failure.

Checkpoint

A checkpoint is a consistent snapshot of:

  1. The current offset in each source partition
  2. The state of all operators

If the pipeline crashes, it restarts from the last successful checkpoint: sources rewind to the saved offsets, state is restored, and processing continues. This gives exactly-once semantics — every record is processed once, no more, no less.

Checkpointing is based on the Chandy-Lamport algorithm (barriers flow through the stream, operators snapshot state when they see a barrier).

Source ──[record][record][barrier]──→ Map ──[barrier]──→ Sink
                                  ↓                    ↓
                            snapshot state        snapshot state
                            snapshot offset       ack checkpoint
Job

A Job is a complete pipeline definition: Source → Operator(s) → Sink. You submit a Job to the runtime, and it starts consuming, processing, and producing.

Architecture

┌──────────────────────────────────────────────────────┐
│                      Job                              │
│                                                      │
│  Source[Kafka]  →  Operator Chain  →  Sink[Postgres] │
│       │                  │                            │
│       │           ┌──────┴──────┐                     │
│       │           │   State     │                     │
│       │           │  Backend    │                     │
│       │           └──────┬──────┘                     │
│       │                  │                            │
│  ┌────┴──────────────────┴────┐                       │
│  │     Checkpoint Manager     │                       │
│  └────────────────────────────┘                       │
└──────────────────────────────────────────────────────┘

Single-process, multi-goroutine. No cluster runtime. Each source partition is consumed by a separate goroutine. Keyed state is partitioned by key across in-memory maps. Checkpointing uses barrier alignment.

Proposed SDK API

package main

import (
    "context"
    "time"

    "mailer"
    "mailer/source"
    "mailer/sink"
    "mailer/window"
)

func main() {
    env := mailer.NewEnv()

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

    kafkaSink := sink.NewKafkaSink(
        sink.KafkaSinkBrokers("localhost:9092"),
        sink.KafkaSinkTopic("order-summary"),
    )

    env.
        FromSource(kafkaSource).
        KeyBy(func(r mailer.Record) []byte { return r.Key }).
        Window(window.Tumbling(5 * time.Minute)).
        Reduce(func(accum, curr mailer.Record) mailer.Record {
            // sum order amounts per customer per 5 minutes
            return merge(accum, curr)
        }).
        ToSink(kafkaSink)

    env.Execute(context.Background())
}
Fluent Chain
stream.
    Map(parseOrder).              // deserialize JSON
    Filter(isValidOrder).         // drop invalid
    KeyBy(orderCustomerKey).      // partition by customer ID
    Window(window.Tumbling(5min)). // 5-minute tumbling window
    Reduce(aggregateAmount).      // sum amounts
    ToSink(kafkaSink)             // write results
ProcessFunction (low-level)
stream.Process(func(ctx mailer.Context, r mailer.Record) {
    state := ctx.ValueState("count")

    count := state.Get() + 1
    state.Set(count)

    if count >= 100 {
        ctx.Output(alertRecord)
        state.Set(0)
    }

    ctx.Collect(r)
})

Package Structure

mailer/
├── mailer.go              # StreamExecutionEnv, NewEnv(), Execute()
├── stream.go              # Stream type (fluent chain builder)
├── record.go              # Record type
├── operator/
│   ├── operator.go        # Operator interface
│   ├── map.go             # Map operator
│   ├── flatmap.go         # FlatMap operator
│   ├── filter.go          # Filter operator
│   ├── keyby.go           # KeyBy partitioner
│   ├── reduce.go          # Reduce operator
│   └── process.go         # ProcessFunction operator
├── window/
│   ├── window.go          # WindowAssigner interface
│   ├── tumbling.go        # Tumbling window
│   ├── sliding.go         # Sliding window
│   └── session.go         # Session window
├── state/
│   ├── state.go           # StateBackend interface, State types
│   └── memory.go          # In-memory state backend
├── source/
│   ├── source.go          # Source interface
│   ├── kafka.go           # Kafka source (using segmentio/kafka-go)
│   ├── kafka_options.go   # Kafka source functional options
│   ├── kafka_offset.go    # Mailer offset enum (no kafka-go leak)
│   ├── deserialize.go     # Deserializer interface + JSON impl
│   ├── watermark.go       # WatermarkSource wrapper
│   └── generator.go       # Test data generator source
├── sink/
│   ├── sink.go            # Sink interface
│   ├── kafka.go           # Kafka sink (options-based, SASL/TLS, serializer)
│   ├── kafka_options.go   # Kafka sink functional options
│   ├── serialize.go       # Serializer interface + JSON impl
│   ├── postgres.go        # Postgres sink (pgx, batch insert, retry)
│   ├── postgres_options.go # Postgres sink functional options
│   ├── stdout.go          # Stdout sink
│   └── blackhole.go       # Blackhole sink (benchmarking)
├── checkpoint/
│   ├── checkpoint.go      # CheckpointManager, barrier injection
│   └── store.go           # Checkpoint storage (file / memory)
├── watermark/
│   └── watermark.go       # Watermark generator strategies
└── examples/
    ├── wordcount/         # Basic map + reduce
    ├── kafka-orders/      # Kafka → Window → Kafka
    └── fraud-detect/      # Stateful process function

Implementation Phases

Phase 1: Core Pipeline (no Kafka, no state, no windows)

Build the foundation — Record, Stream, basic operators, Source/Sink interfaces, in-memory sources and sinks.

What works: GeneratorSource → Map → Filter → StdoutSink

File What it does
record.go Record type with Key, Value, Timestamp, Offset
mailer.go StreamExecutionEnv, FromSource(), Execute()
stream.go Stream chain builder — Map(), Filter(), KeyBy(), ToSink()
operator/operator.go Operator interface
operator/map.go Map operator implementation
operator/filter.go Filter operator implementation
operator/keyby.go KeyBy partitioner
source/source.go Source interface
source/generator.go Generator source for testing
sink/sink.go Sink interface
sink/stdout.go Stdout sink
Phase 2: Stateful Processing

Add per-key state and the Reduce operator. State lives in-memory, partitioned by key after KeyBy.

What works: Source → KeyBy → Reduce(accumulator) → Sink

File What it does
state/state.go StateBackend interface, ValueState, ListState
state/memory.go In-memory state backend (map per key)
operator/reduce.go Reduce operator with keyed state
operator/process.go ProcessFunction with state access
Phase 3: Windowing & Watermarks

Add time-based windowing. Watermarks flow through the stream and trigger window completion.

What works: Source → KeyBy → Window(5min) → Reduce → Sink

File What it does
watermark/watermark.go Watermark generator (bounded out of orderness)
window/window.go WindowAssigner interface, WindowResult
window/tumbling.go Tumbling window
window/sliding.go Sliding window
window/session.go Session window
Phase 4: Kafka Source & Sink

Connect to real Kafka. Consume topics with consumer groups, produce to topics, commit offsets.

What works: KafkaSource → Window → Reduce → KafkaSink

File What it does
source/kafka.go Kafka source using segmentio/kafka-go
sink/kafka.go Kafka sink
sink/blackhole.go Blackhole sink for benchmarking
Phase 5: Checkpointing & Fault Tolerance

Add barrier-based checkpointing. On failure, restore state and rewind source offsets.

What works: Exactly-once processing with crash recovery.

File What it does
checkpoint/checkpoint.go Barrier injection, snapshot coordination
checkpoint/store.go Checkpoint storage (filesystem / in-memory)

Key Design Decisions

Why single-process, not distributed?

Flink runs as a JobManager + TaskManager cluster. Mailer runs as a single Go process with goroutines. This makes it simple, embeddable, and easy to reason about. If you need multi-node parallelism, run multiple Mailer instances with different consumer group IDs (Kafka handles partitioning).

Why barrier-based checkpointing?

The Chandy-Lamport approach (barriers flow through the dataflow graph, operators snapshot on barrier arrival) is well-proven in Flink. It provides exactly-once semantics without stopping the stream. The barrier is a special Record that flows in-band with data.

Why not just use Kafka Streams?

Kafka Streams is Java-only. Mailer gives Go developers a native, embeddable stream processing library with similar semantics — without the JVM.

Why segmentio/kafka-go over confluent-kafka-go?

segmentio/kafka-go is pure Go (no CGO), which simplifies builds and cross-compilation. If performance becomes critical, we can add a confluent-kafka-go source as an alternative.

Feature Apache Flink Mailer
Language Java Go
Deployment Cluster (JobManager + TaskManagers) Single process, embeddable
API DataStream API, Table API, SQL DataStream API (fluent chain)
State Backends RocksDB, Memory Memory (v1), pluggable
Checkpointing Barrier-based, aligned/unaligned Barrier-based, aligned
Windowing Tumbling, Sliding, Session, Global Tumbling, Sliding, Session
Kafka Connector Built-in, mature Built-in (segmentio/kafka-go)
Exactly-once Yes Yes (via checkpointing)
SQL Yes No (not planned for v1)
CEP (Complex Event Processing) Yes No (future)

Status

Pre-development. This README defines the architecture and implementation plan. Code starts from Phase 1.

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type CheckpointInfo 

type CheckpointInfo struct {
	Interval string `json:"interval"`
	Storage  string `json:"storage"`
}

CheckpointInfo describes the checkpointing configuration.

type OperatorInfo 

type OperatorInfo struct {
	Type   string            `json:"type"`
	Label  string            `json:"label,omitempty"`
	Config map[string]string `json:"config,omitempty"`
}

OperatorInfo describes a single operator in the pipeline chain.

type PipelineInfo 

type PipelineInfo struct {
	Source     source.SourceInfo `json:"source"`
	Operators  []OperatorInfo    `json:"operators"`
	Sink       sink.SinkInfo     `json:"sink"`
	Checkpoint *CheckpointInfo   `json:"checkpoint,omitempty"`
}

PipelineInfo is the JSON-serializable description of a pipeline. It is returned by Describe() and consumed by the dashboard.

type Stream 

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

Stream represents a pipeline stage. Method calls on Stream append operators to the chain and return the updated Stream.

Streams are built using a fluent API: 

env.FromSource(src).Map(fn).Filter(fn).ToSink(s)

The pipeline is lazy — nothing runs until env.Execute() is called.

func (*Stream) Filter 

func (s *Stream) Filter(fn func(types.Record) bool, label ...string) *Stream

Filter keeps only records where fn returns true. The optional label is shown in the dashboard.

func (*Stream) FlatMap 

func (s *Stream) FlatMap(fn func(types.Record) []types.Record, label ...string) *Stream

FlatMap applies a 1:many transformation to each record. The function returns a slice; if empty, the record is dropped. The optional label is shown in the dashboard.

func (*Stream) KeyBy 

func (s *Stream) KeyBy(fn func(types.Record) []byte, label ...string) *Stream

KeyBy partitions the stream by the given key selector function. All records with the same key are routed together. Required before stateful operations like Reduce. The optional label is shown in the dashboard.

func (*Stream) Map 

func (s *Stream) Map(fn func(types.Record) types.Record, label ...string) *Stream

Map applies a 1:1 transformation to each record in the stream. If the function returns the zero value of Record, the record is dropped. The optional label is shown in the dashboard.

func (*Stream) Reduce 

func (s *Stream) Reduce(fn operator.ReduceFn, label ...string) *Stream

Reduce applies a stateful aggregation per key. Must be used after KeyBy. The reduce function is called with the current accumulator (nil on first call) and the incoming record, and returns the new accumulator. The updated accumulator is emitted downstream after every record.

Example (count per key): 

stream.KeyBy(func(r types.Record) []byte { return r.Key }).
    Reduce(func(accum []byte, curr types.Record) []byte {
        count := 0
        if accum != nil {
            count = int(binary.BigEndian.Uint64(accum))
        }
        count++
        buf := make([]byte, 8)
        binary.BigEndian.PutUint64(buf, uint64(count))
        return buf
    })

func (*Stream) ToSink 

func (st *Stream) ToSink(sk sink.Sink) *StreamExecutionEnv

ToSink connects the stream to a sink and returns the execution environment. The pipeline is still lazy — call env.Execute() to start processing.

func (*Stream) Window 

func (s *Stream) Window(assigner window.WindowAssigner, label ...string) *Stream

Window groups records into time-based windows. Must be used after KeyBy. Records are buffered into windows, and when a watermark passes a window's end time, the window fires — all its records are emitted as a single result.

Supported window types:

  • window.Tumbling(size): fixed-size, non-overlapping (e.g. 5-minute buckets)
  • window.Sliding(size, slide): overlapping windows (e.g. 5-min every 1-min)
  • window.Session(gap): variable-size, closes after inactivity gap

Example (5-minute tumbling window): 

stream.KeyBy(func(r types.Record) []byte { return r.Key }).
    Window(window.Tumbling(5 * time.Minute)).
    Reduce(aggregateFn)

func (*Stream) WindowWithIdleTimeout 

func (s *Stream) WindowWithIdleTimeout(assigner window.WindowAssigner, idleTimeout time.Duration, label ...string) *Stream

WindowWithIdleTimeout creates a window with an idle timeout. If no records arrive within the timeout duration, all remaining windows are fired and the pipeline stage completes. Useful for infinite streams that don't receive shutdown signals.

type StreamExecutionEnv 

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

StreamExecutionEnv is the entry point for building and running stream pipelines. Create one with NewEnv(), define your pipeline using FromSource/ToSink, then call Execute() to run it. 

env := mailer.NewEnv()
env.FromSource(src).Map(fn).Filter(fn).ToSink(stdout)
env.Execute(ctx)

func NewEnv 

func NewEnv() *StreamExecutionEnv

NewEnv creates a new StreamExecutionEnv.

func (*StreamExecutionEnv) Describe 

func (env *StreamExecutionEnv) Describe() PipelineInfo

Describe returns a serializable description of the pipeline: source, operator chain, sink, and checkpoint config. Call this before Execute() to get the pipeline graph for the dashboard.

func (*StreamExecutionEnv) DescribeJSON 

func (env *StreamExecutionEnv) DescribeJSON() string

DescribeJSON returns the pipeline description as indented JSON.

func (*StreamExecutionEnv) Execute 

func (env *StreamExecutionEnv) Execute(ctx context.Context) error

Execute runs the pipeline. It starts the source, wires up all operators as goroutines connected by channels, and connects the final output to the sink. Blocks until the source is exhausted or the context is cancelled.

If checkpointing is enabled, the pipeline will attempt to restore from the latest checkpoint before starting. A goroutine injects checkpoint barriers at the configured interval. When a barrier completes the full pipeline round-trip, operator state is saved to the checkpoint storage.

func (*StreamExecutionEnv) FromSource 

func (env *StreamExecutionEnv) FromSource(src source.Source) *Stream

FromSource sets the data source for the pipeline and returns a Stream that you can chain operators on.

func (*StreamExecutionEnv) WithCheckpointing 

func (env *StreamExecutionEnv) WithCheckpointing(interval time.Duration, storage checkpoint.Storage) *StreamExecutionEnv

WithCheckpointing enables periodic checkpointing with the given interval and storage backend. Barriers are injected into the stream at the specified interval; when a barrier passes through all operators and reaches the sink, the checkpoint is complete.

On recovery, Execute() will load the latest checkpoint, restore all stateful operators, and resume from the saved source offset.

Example: 

env := mailer.NewEnv()
env.WithCheckpointing(30*time.Second, checkpoint.NewFileStorage("/tmp/checkpoints"))

Source Files

View all Source files

Directories

Path Synopsis
control module
examples
dashboard-demo command
kafka-dashboard command
kafka-orders command
pg-orders command
windowing command
wordcount command
Package state provides keyed state storage for stateful stream processing.
 Package state provides keyed state storage for stateful stream processing.
Package watermark provides watermark generation for event-time processing.
 Package watermark provides watermark generation for event-time processing.
Package window provides window assigners for grouping unbounded streams into finite chunks based on time.
 Package window provides window assigners for grouping unbounded streams into finite chunks based on time.

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.