kcl-go

Production-grade Go library for Amazon Kinesis Data Streams — full KCL parity, pluggable lease storage, and clean middleware-based observability.

Features

Full KCL parity — polling consumer, Enhanced Fan-Out (EFO), DynamoDB lease coordination, at-least-once checkpointing, automatic resharding
Pluggable lease backends — DynamoDB (default), PostgreSQL, or bring your own pkg/lease.Store implementation
KPL aggregation — WithKPLAggregation() packs multiple records into a single Kinesis record; DeaggregateRecords transparently expands on the consumer side
Transparent compression — WithCompression / WithDecompression with gzip, zstd, or lz4; native wire format
Middleware pattern — pkg/middleware.Chain decouples business logic from logging and metrics
Functional options — clean configuration without a builder struct
Production defaults — 80% coverage gate, race detector, golangci-lint, GOTOOLCHAIN=local

Why This Library? Comparison

vs. Java KCL (amazon-kinesis-client)

The Java KCL is the reference implementation. Go teams typically had two options: run the Java KCL as a sidecar (JVM overhead, cross-process complexity) or use the Multilanguage Daemon (MLD) which still requires a JVM process on the same host.

Capability	Java KCL v2	kcl-go
Language	Java	Go — no JVM
Deployment model	Standalone process or sidecar	Importable Go library
Memory footprint	~200–400 MB (JVM heap)	~10–30 MB
Cold-start time	2–10s (JVM startup)	<100ms
Enhanced Fan-Out (EFO)	✅	✅
Polling consumer	✅	✅
DynamoDB lease coordination	✅	✅
At-least-once checkpointing	✅	✅
Automatic resharding	✅	✅
Multi-worker distributed leases	✅	✅
Multi-stream consumer (KCL 2.4+)	✅	⚠️ Not yet — see Roadmap
Pluggable lease storage	❌ DynamoDB only	✅ DynamoDB, PostgreSQL, or custom
Transparent compression	❌	✅ gzip / zstd / lz4 — `WithCompression`
KPL producer aggregation	✅ (via Java KPL library)	✅ `WithKPLAggregation()` — native Go
Observability	Metrics via CloudWatch	✅ Victoria Metrics, CloudWatch, or custom
Middleware / DI pattern	❌ coupled	✅ middleware.Chain
Functional options API	❌ builder	✅ WithXxx options
Go-native context propagation	❌	✅
Integration test harness	Manual / TestNG	✅ testcontainers-go + LocalStack

vs. Raw AWS SDK Go v2

Using aws-sdk-go-v2/service/kinesis directly means implementing all coordination from scratch.

Capability	Raw SDK (`aws-sdk-go-v2`)	kcl-go
Records delivery	Manual `GetRecords` loop	✅ Managed
Shard discovery	Manual `ListShards` + pagination	✅ Managed
Multi-shard parallelism	Manual goroutines	✅ One goroutine per shard
Distributed leases	❌ DIY DynamoDB	✅ Built-in
Checkpointing	❌ DIY	✅ Built-in
Lease expiry + steal	❌ DIY	✅ Built-in
Resharding detection	❌ DIY	✅ Built-in
Enhanced Fan-Out	Manual HTTP/2 subscription	✅ Managed reconnect
Transparent compression	❌ DIY	✅ `WithCompression` / `WithDecompression`
KPL producer aggregation	❌ DIY	✅ `WithKPLAggregation()`
Graceful shutdown	DIY drain logic	✅ `context.Cancel` propagation
Observability	DIY	✅ Middleware layer

Rule of thumb: if you need more than one shard or more than one process, use this library instead of raw SDK.

vs. Other Go Kinesis Libraries

Library	Stars	Maintained	EFO	Distributed	Lease Backends	Compression	KPL Aggregation
`harlow/kinesis-consumer`	~600	Archived 2021	❌	❌ single-process	❌	❌	❌
`vmware/vmware-go-kcl`	~300	Archived 2023	❌	✅ DynamoDB	DynamoDB only	❌	❌ consumer only
`aws-sdk-go-v2` (raw)	Official	✅ Active	Manual	❌ DIY	❌ DIY	❌ DIY	❌ DIY
kcl-go	This repo	✅ Active	✅	✅ Multi-process	DynamoDB, PostgreSQL, custom	✅ gzip / zstd / lz4	✅ `WithKPLAggregation()`

Key differentiators of this library:

Full EFO support — no other active Go library implements SubscribeToShard with reconnect
KPL aggregation on producer — WithKPLAggregation() in native Go; no Java KPL sidecar needed
Pluggable lease storage — swap DynamoDB for PostgreSQL without changing consumer code
Transparent compression — gzip / zstd / lz4 on producer and consumer; native wire format; third-party compatible
Middleware-first observability — Victoria Metrics and CloudWatch are layers, not embedded
Production test coverage — 80% gate, race detector, integration tests via LocalStack

Installation

go get github.com/senthilsam/kcl-go

Quick Start: Polling Consumer

package main

import (
    "context"
    "log/slog"
    "os/signal"
    "syscall"

    awsconfig "github.com/aws/aws-sdk-go-v2/config"
    "github.com/senthilsam/kcl-go/pkg/consumer"
    pkglogger "github.com/senthilsam/kcl-go/pkg/logger"
    pkgmetrics "github.com/senthilsam/kcl-go/pkg/metrics"
    "github.com/senthilsam/kcl-go/pkg/middleware"
)

// myProcessor handles one batch of records per call.
type myProcessor struct{ consumer.NoopRecordProcessor }

func (p *myProcessor) ProcessRecords(ctx context.Context, in consumer.ProcessRecordsInput) error {
    for _, r := range in.Records {
        slog.Info("record", "seq", *r.SequenceNumber, "bytes", len(r.Data))
    }
    if len(in.Records) > 0 {
        return in.Checkpointer.Checkpoint(ctx, *in.Records[len(in.Records)-1].SequenceNumber)
    }
    return nil
}

func main() {
    awsCfg, _ := awsconfig.LoadDefaultConfig(context.Background())
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM)
    defer stop()

    base, err := consumer.New(consumer.Config{
        StreamName:             "my-stream",
        ApplicationName:        "my-app",
        AWSConfig:              awsCfg,
        Processor:              &myProcessor{},
        CreateTableIfNotExists: true,
    })
    if err != nil {
        slog.Error("create consumer", "err", err)
        return
    }

    // Wrap with logging + metrics middleware (business logic stays clean).
    c := middleware.Chain(base,
        consumer.LoggingMiddleware(pkglogger.NewSlogAdapter(slog.Default())),
        consumer.MetricsMiddleware("kinesis_consumer", &pkgmetrics.NoopMetricsClient{}),
    )
    c.Start(ctx)
}

Quick Start: Enhanced Fan-Out (EFO)

base, _ := consumer.New(
    consumer.Config{
        StreamName:      "my-stream",
        ApplicationName: "my-app-efo",
        Processor:       &myProcessor{},
    },
    consumer.WithEnhancedFanOut(),
    consumer.WithCreateTableIfNotExists(),
)

Quick Start: Producer

p, err := producer.New(producer.Config{
    StreamName: "my-stream",
    AWSConfig:  cfg,
}, producer.WithMaxRetries(3))
defer p.Close(ctx)

// Buffered single record (auto-batched):
p.Put(ctx, "partition-key", []byte(`{"event":"click"}`))

// Explicit batch of up to 500:
results, err := p.PutBatch(ctx, []producer.ProducerRecord{
    {PartitionKey: "pk-1", Data: []byte(`{"id":1}`)},
    {PartitionKey: "pk-2", Data: []byte(`{"id":2}`)},
})

Configuration Reference

Consumer

Option	Default	Description
`StreamName`	required	Kinesis stream name
`ApplicationName`	required	Consumer group name / DynamoDB table name
`Processor`	required	`RecordProcessor` implementation
`WithWorkerID(id)`	hostname+PID	Unique identifier for lease coordination
`WithPollingInterval(d)`	1s	Interval between `GetRecords` calls
`WithHeartbeatInterval(d)`	3s	Lease heartbeat frequency
`WithLeaseExpiryDuration(d)`	10s	Lease expiry timeout
`WithShardSyncInterval(d)`	60s	Shard re-discovery frequency
`WithMaxRecordsPerCall(n)`	10000	Records per `GetRecords` call
`WithEnhancedFanOut()`	false	Use EFO (`SubscribeToShard`)
`WithCreateTableIfNotExists()`	false	Auto-create lease table
`WithLeaseStore(s)`	DynamoDB	Custom lease backend
`WithLogger(l)`	`slog.Default()`	Structured logger
`WithMetrics(m)`	NoopMetricsClient	Metrics backend

Producer

Option	Default	Description
`StreamName`	required	Kinesis stream name
`AWSConfig`	required	AWS SDK config
`WithMaxRetries(n)`	3	Per-record retry count
`WithBatchSize(n)`	500	Max records per `PutRecords`
`WithBatchBytes(n)`	5 MB	Max payload per `PutRecords`
`WithBufferSize(n)`	500	Internal send buffer slots
`WithProducerLogger(l)`	`slog.Default()`	Structured logger
`WithProducerMetrics(m)`	NoopMetricsClient	Metrics backend
`WithEFOSafePartitionKeys()`	false	Reject mixed-PK batches (see EFO Safety)

KPL Aggregated Records

The Kinesis Producer Library (KPL) can aggregate multiple user records into a single Kinesis record to improve throughput efficiency and reduce per-record costs. This is transparent in the Java KCL — the library automatically expands them before delivery.

This library supports KPL aggregation via the pkg/aggregation package.

How It Works

┌─────────────────────────────────────────────────────────────┐
│  Kinesis Record (single PutRecords entry)                   │
│  ┌──────────┬──────────────────────────────┬────────────┐  │
│  │ 4 bytes  │  protobuf AggregatedRecord   │  16 bytes  │  │
│  │  magic   │  (partition keys + records)  │    MD5     │  │
│  └──────────┴──────────────────────────────┴────────────┘  │
│                           ↓ DeaggregateRecords                │
│  UserRecord{Data: payload1, SubSequenceNumber: 0}             │
│  UserRecord{Data: payload2, SubSequenceNumber: 1}             │
│  UserRecord{Data: payload3, SubSequenceNumber: 2}             │
└─────────────────────────────────────────────────────────────┘

Magic bytes: 0xF3 0x89 0x9A 0xC2 — detected by aggregation.IsAggregated()
Protobuf payload: AggregatedRecord proto2 message (hand-parsed — no protobuf dependency)
MD5 checksum: 16 bytes appended, verified during de-aggregation
SubSequenceNumber: 0-based index of each user record within the aggregate — used for checkpointing position within the batch

Usage

Call DeaggregateRecords inside ProcessRecords for transparent KPL support:

import "github.com/senthilsam/kcl-go/pkg/aggregation"

func (p *myProcessor) ProcessRecords(ctx context.Context, records []ktypes.Record, c checkpoint.Checkpointer) error {
    // Expand any KPL-aggregated records into individual user records.
    // Non-aggregated records pass through unchanged (SubSequenceNumber = 0).
    expanded, err := aggregation.DeaggregateRecords(records)
    if err != nil {
        return err
    }

    for _, ur := range expanded {
        // ur.Record.Data  — original user payload
        // ur.Aggregated   — true if came from a KPL aggregate
        // ur.SubSequenceNumber — position within aggregate (0 for plain records)
        slog.Info("user record",
            "seq", *ur.Record.SequenceNumber,
            "sub_seq", ur.SubSequenceNumber,
            "bytes", len(ur.Record.Data),
        )
    }

    // Checkpoint at last user record (sub-sequence aware).
    if len(expanded) > 0 {
        last := expanded[len(expanded)-1]
        return c.CheckpointAt(ctx, *last.Record.SequenceNumber, last.SubSequenceNumber)
    }
    return nil
}

Why SubSequenceNumber Matters

Without sub-sequence checkpointing, a restart after processing user record #47 of a 100-record aggregate would re-process records 0–47. With CheckpointAt, the restart resumes at #48 — true at-least-once semantics at user-record granularity.

Method	When to use
`c.Checkpoint(ctx, seqNum)`	Standard records, or when you don't care about intra-aggregate position
`c.CheckpointAt(ctx, seqNum, subSeqNum)`	KPL aggregated records — use `ur.SubSequenceNumber`

Producer Aggregation

This library supports KPL-format aggregation on the produce side via WithKPLAggregation(). Multiple user records are packed into a single Kinesis record payload (up to 1 MB each), reducing per-record API costs by up to 1000× and increasing throughput significantly for small-payload workloads.

p, err := producer.New(producer.Config{
    StreamName: "my-stream",
    AWSConfig:  cfg,
}, producer.WithKPLAggregation())

// Combine with compression for maximum efficiency:
p, err := producer.New(cfg,
    producer.WithKPLAggregation(),             // pack N records into 1 Kinesis record
    producer.WithCompression(compression.Gzip), // compress the aggregated payload
)

The consumer side is unchanged — call DeaggregateRecords in ProcessRecords as shown above. The outer partition key of each aggregated Kinesis record is set to the first inner record's partition key (Lambda EFO safety — prevents silent drops by Lambda's EFO consumer).

Concern	Behavior
Payload limit	Packs records until the next would exceed 1 MB; starts a new payload
Outer partition key	First inner record's partition key
Lambda EFO drops	Avoided by outer PK = first inner PK; combine with `WithEFOSafePartitionKeys()` for strict enforcement
Consumer compatibility	Java KCL, kcl-go (`DeaggregateRecords`), Lambda EFO
Wire format	`0xF3 0x89 0x9A 0xC2` + protobuf + MD5 (standard KPL format)

If you only use this library's producer without WithKPLAggregation(), de-aggregation is a no-op (all records have Aggregated: false, SubSequenceNumber: 0).

EFO Safety and Partition Key Enforcement

Important for Lambda EFO consumers with upstream KPL aggregators

AWS Lambda's Enhanced Fan-Out (EFO) consumer via SubscribeToShard silently drops any inner KPL user record whose partition key does not match the outer Kinesis record's partition key. This is a Lambda-specific behavior — this library's own consumer always delivers all records regardless of partition key mismatch.

If all three of these apply to your system:

Records are written by the Java KPL (or compatible aggregator)
The Kinesis stream is consumed by AWS Lambda with EFO
Multiple records with different partition keys may end up in the same aggregate

…then Lambda will silently drop some inner records. To prevent this, enable EFOSafePartitionKeys on the producer:

p, err := producer.New(cfg,
    producer.WithEFOSafePartitionKeys(), // reject mixed-PK batches at the producer
)

When enabled, PutBatch returns ErrPartitionKeyMismatch if the batch contains records with different partition keys. Split the batch by partition key before calling PutBatch:

// Group records by partition key before sending.
byPK := make(map[string][]producer.ProducerRecord)
for _, r := range records {
    byPK[r.PartitionKey] = append(byPK[r.PartitionKey], r)
}
for _, group := range byPK {
    if _, err := p.PutBatch(ctx, group); err != nil {
        // handle
    }
}

Scenario	Behavior
`EFOSafePartitionKeys=false` (default)	All records sent as-is; Lambda EFO may silently drop mismatched inner records
`EFOSafePartitionKeys=true`, same PK	Allowed — batch proceeds normally
`EFOSafePartitionKeys=true`, mixed PKs	Returns `ErrPartitionKeyMismatch` — split before send
This library's consumer (any mode)	Always delivers all records; partition key mismatch has no effect

// PostgreSQL:
import "github.com/senthilsam/kcl-go/internal/lease"
import _ "github.com/jackc/pgx/v5/stdlib"

db, _ := sql.Open("pgx", os.Getenv("DATABASE_URL"))
pgStore := lease.NewPostgresStore(db, "kinesis_leases", logger)

c, _ := consumer.New(cfg, consumer.WithLeaseStore(pgStore))

// Custom backend — implement pkg/lease.Store:
type myStore struct{}
func (s *myStore) Setup(ctx) error                                                 { ... }
func (s *myStore) CreateIfNotExists(ctx, shardID) error                           { ... }
// ... 7 methods total
c, _ := consumer.New(cfg, consumer.WithLeaseStore(&myStore{}))

Middleware

Business logic stays clean. Observability is layered on via middleware.Chain:

import (
    "log/slog"
    internmetrics "github.com/senthilsam/kcl-go/internal/metrics"
    pkglogger      "github.com/senthilsam/kcl-go/pkg/logger"
    "github.com/senthilsam/kcl-go/pkg/middleware"
)

log := pkglogger.NewSlogAdapter(slog.Default())
vm  := internmetrics.NewVictoriaMetricsClient("kinesis_consumer")

base, _ := consumer.New(cfg)
c := middleware.Chain(base,
    consumer.LoggingMiddleware(log),            // inner — logs Start/Stop
    consumer.MetricsMiddleware("kinesis_consumer", vm), // outer — emits {ns}_up, {ns}_uptime_ms
)

pvm  := internmetrics.NewVictoriaMetricsClient("kinesis_producer")
plog := pkglogger.NewSlogAdapter(slog.Default())
p, _ := producer.New(pcfg)
p = middleware.Chain(p,
    producer.LoggingMiddleware(plog),
    producer.MetricsMiddleware("kinesis_producer", pvm),
)

Compression

Records can be compressed transparently — the producer compresses before sending, the consumer decompresses before records reach your ProcessRecords implementation. Business logic is completely unaware of compression.

Three codecs are available:

Codec	Constructor	Typical ratio	CPU cost	Wire format
gzip	`compression.Gzip`	2–4×	Low	RFC 1952
zstd	`compression.Zstd`	3–5×	Medium	RFC 8878
lz4	`compression.Lz4`	1.5–3×	Very low	LZ4 frame

Each codec stores records in its native wire format — any compatible library can decompress without knowing about kcl-go.

Producer — enable compression

p, err := producer.New(producer.Config{
    StreamName: "my-stream",
    AWSConfig:  cfg,
}, producer.WithCompression(compression.Gzip))
// also: compression.Zstd, compression.Lz4

The 5 MB / 500-record batch limits are enforced on compressed bytes — a batch that compresses 20 MB of originals to 4 MB fits in a single PutRecords call.

Consumer — enable decompression

c, err := consumer.New(consumer.Config{
    StreamName:      "my-stream",
    ApplicationName: "my-app",
    AWSConfig:       cfg,
    Processor:       &myProcessor{}, // receives original uncompressed bytes
}, consumer.WithDecompression(compression.Gzip))

Records that don't match the codec's magic bytes pass through unchanged — safe for streams with mixed compressed and uncompressed producers.

Combining with middleware

Compression/decompression compose with the existing middleware pattern without any changes:

// Consumer: decompression option + Consumer-level middleware
base, _ := consumer.New(
    consumer.Config{Processor: &myProcessor{}},
    consumer.WithDecompression(compression.Gzip),
)
c := middleware.Chain(base,
    consumer.LoggingMiddleware(log),
    consumer.MetricsMiddleware("kinesis_consumer", vm),
)

// Or use RecordProcessor-level middleware for per-batch observability:
cfg.Processor = middleware.Chain[consumer.RecordProcessor](
    &myProcessor{},
    consumer.LoggingProcessorMiddleware(log),
    consumer.MetricsProcessorMiddleware("kinesis", vm),
    consumer.DecompressionMiddleware(compression.Gzip),
)

Advanced: BatchTransformer chain (producer)

For custom transform pipelines — compression, encryption, schema validation, etc.:

p, _ := producer.New(producer.Config{
    StreamName: "my-stream",
    Transformer: middleware.Chain[producer.BatchTransformer](
        producer.IdentityTransformer(),
        producer.CompressionTransformerMiddleware(compression.Gzip),
        producer.LoggingTransformerMiddleware(log),
        producer.MetricsTransformerMiddleware("kinesis_producer", vm),
    ),
})

Security: decompression bomb protection

WithDecompression enforces a 10 MB limit per record by default:

consumer.WithDecompression(compression.Gzip,
    consumer.WithMaxDecompressedSize(5*1024*1024), // override to 5 MB
)

Records exceeding the limit return compression.ErrDecompressedSizeLimitExceeded from ProcessRecords. The size limit is checked before allocating the full decompression buffer.

See the compressed producer-consumer example.

Observability

Metric Backends

The library uses a pluggable MetricsClient interface (pkg/metrics). Three implementations are provided out of the box in internal/metrics:

Backend	Constructor	Notes
Victoria Metrics (Prometheus)	`metrics.NewVictoriaMetricsClient(namespace)`	Pull-model, exposes `/metrics` HTTP endpoint
Amazon CloudWatch	`metrics.NewCloudWatchClient(cwClient, cwNamespace, prefix, logger)`	Push-model, batched flush every 60 s
Multi-backend fanout	`metrics.NewMultiClient(vm, cw, ...)`	Writes to all backends simultaneously
No-op (default)	`&pkgmetrics.NoopMetricsClient{}`	Discards all metrics — zero overhead

Import path: github.com/senthilsam/kcl-go/internal/metrics

Victoria Metrics (Prometheus)

import (
    internmetrics "github.com/senthilsam/kcl-go/internal/metrics"
    "net/http"
)

vm := internmetrics.NewVictoriaMetricsClient("kinesis_consumer")

// Wire into consumer
c, _ := consumer.New(cfg, consumer.WithMetrics(vm))

// Wire into producer
p, _ := producer.New(pcfg, producer.WithProducerMetrics(vm))

// Expose /metrics endpoint (Prometheus scrape target)
http.HandleFunc("/metrics", vm.HTTPHandler())
http.ListenAndServe(":2112", nil)

Metric names follow the Prometheus convention {namespace}_{metric_name}. Counters use the _total suffix. Example: kinesis_consumer_records_processed_total{stream="my-stream",shard="shardId-000000000000"}.

Amazon CloudWatch

import (
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/cloudwatch"
    internmetrics "github.com/senthilsam/kcl-go/internal/metrics"
)

awsCfg, _ := config.LoadDefaultConfig(ctx)
cwClient  := cloudwatch.NewFromConfig(awsCfg)

cw := internmetrics.NewCloudWatchClient(
    cwClient,
    "KinesisClientGo/Consumer", // CloudWatch namespace
    "kinesis_consumer",         // metric name prefix
    nil,                        // logger — nil uses slog.Default()
)
defer cw.Stop() // flush remaining metrics on shutdown

c, _ := consumer.New(cfg, consumer.WithMetrics(cw))

Metrics are buffered and flushed every 60 seconds or when the buffer reaches 1 000 data points (the CloudWatch PutMetricData limit). Call cw.Stop() before your process exits to flush the final batch.

Both Backends Simultaneously

vm := internmetrics.NewVictoriaMetricsClient("kinesis_consumer")
cw := internmetrics.NewCloudWatchClient(cwClient, "KinesisClientGo/Consumer", "kinesis_consumer", nil)

multi := internmetrics.NewMultiClient(vm, cw) // fans out to both

c, _ := consumer.New(cfg, consumer.WithMetrics(multi))
p, _ := producer.New(pcfg, producer.WithProducerMetrics(multi))

Custom Backend

Implement pkg/metrics.MetricsClient to route metrics to any backend (StatsD, Datadog, OpenTelemetry, etc.):

type myMetrics struct{}

func (m *myMetrics) PutMetric(ctx context.Context, name string, value float64, unit string, dims []metrics.Dimension) error {
    // forward to your backend
    return nil
}

c, _ := consumer.New(cfg, consumer.WithMetrics(&myMetrics{}))

Consumer Metric Reference

All metrics carry stream and shard dimensions unless noted.

Metric	Type	Description	When emitted
`millis_behind_latest`	gauge	Consumer lag — ms the shard is behind the tip of the stream. The primary alert metric.	Every poll / EFO event
`records_processed_total`	counter	Records successfully delivered to `ProcessRecords`	Per successful batch
`bytes_processed_total`	counter	Payload bytes delivered (excludes partition keys)	Per successful batch
`process_duration_ms`	gauge	`ProcessRecords` wall-clock latency	Per batch
`fetch_duration_ms`	gauge	`GetRecords` API round-trip (polling only)	Per poll
`empty_polls_total`	counter	Polls that returned 0 records (polling only) — high values indicate over-polling	Per empty poll
`iterator_refreshes_total`	counter	Expired iterator recoveries (polling only)	On iterator expiry
`fetch_errors_total`	counter	`GetRecords` transport errors (polling only)	On fetch failure
`process_errors_total`	counter	`ProcessRecords` failures after all retries	On exhausted retry
`lease_lost_total`	counter	Lease loss events — triggers shard re-assignment	On lease loss
`efo_reconnects_total`	counter	EFO stream reconnect attempts (EFO only)	On subscribe failure or stream error
`efo_stream_errors_total`	counter	EFO stream closed with error (EFO only)	On stream close
`{ns}_up`	gauge	1 while consumer is running, 0 after Stop	On Start / Stop
`{ns}_uptime_ms`	gauge	Total consumer uptime per run (dims: `result={success\|error}`)	On Stop

Producer Metric Reference

All metrics carry a stream dimension.

Metric	Type	Description
`records_sent_total`	counter	Records successfully written to Kinesis
`bytes_sent_total`	counter	Payload bytes written (first-attempt successes)
`put_latency_ms`	gauge	`PutRecords` API round-trip latency per call
`records_retried_total`	counter	Records queued for retry after per-record failure
`records_failed_total`	counter	Records that exhausted all retry attempts
`put_errors_total`	counter	Fatal transport-level `PutRecords` errors
`buffer_utilization`	gauge	Internal buffer fill level (0–100 %) — spike indicates back-pressure

Recommended Alerts

# Consumer falling behind — SLO alert
millis_behind_latest{stream="my-stream"} > 30000   # > 30 s lag

# Processing errors rising
rate(process_errors_total[5m]) > 0

# EFO instability
rate(efo_reconnects_total[5m]) > 1

# Producer back-pressure
buffer_utilization > 80

# Producer losing records
rate(records_failed_total[5m]) > 0

Deployment Notes

ECS / Kubernetes

Use an IAM role attached to the task/pod — no credentials in code
Set CreateTableIfNotExists: true on first startup
LeaseExpiryDuration default (10s) works well; tune HeartbeatInterval < expiry

AWS Lambda

Use polling mode (EFO is not recommended for Lambda)
Set a short PollingInterval (500ms) and ShardSyncInterval (30s)
Checkpoint after every batch — Lambda may be stopped between invocations

Multi-worker (distributed)

Each process has a unique WorkerID (default: hostname+PID is sufficient for ECS/K8s)
Leases are balanced to ceil(shards/workers) per worker automatically
A crashed worker's leases expire after LeaseExpiryDuration and are stolen

Roadmap

Feature	Status	Notes
Multi-stream consumer (KCL 2.4+)	Planned	Consume multiple Kinesis streams with a single consumer group and shared lease table
`PreparedCheckpointer`	Planned	Two-phase commit — prepare a checkpoint, commit it later
Worker lifecycle hooks	Planned	`WorkerStateChangeListener` equivalent for external service registry integration

Multi-stream support will allow a single consumer.New(...) to process records from multiple streams simultaneously, with leases distributed across all shards from all streams. Target API:

// Future multi-stream API (not yet implemented):
c, err := consumer.NewMultiStream(consumer.MultiStreamConfig{
    Streams: []consumer.StreamConfig{
        {StreamName: "orders-stream",   ApplicationName: "my-app"},
        {StreamName: "payments-stream", ApplicationName: "my-app"},
    },
    Processor: &myProcessor{},
})

Testing

# Unit tests (no Docker required):
task test
# or: GOTOOLCHAIN=local go test ./... -race -coverprofile=coverage.out

# Integration tests (requires Docker):
task test:integration
# or: INTEGRATION=1 go test ./test/... -race -tags integration -timeout 15m

# Per-suite integration tests:
task test:integration:consumer
task test:integration:producer
task test:integration:lease
task test:integration:e2e

# Against an already-running LocalStack (no Docker spin-up):
task test:integration:localstack

# Coverage report:
task coverage:html

Developer Guide

Prerequisites

Tool	Version	Install
Go	1.21+	`brew install go`
Task	3.x	`brew install go-task`
Docker	20.10+	Required for integration tests (LocalStack)
golangci-lint	latest	`brew install golangci-lint`
delve (dlv)	latest	`go install github.com/go-delve/delve/cmd/dlv@latest`

First-time setup

# 1. Tidy dependencies:
task tidy

# 2. Verify everything builds:
task build

# 3. Run unit tests:
task test

Running examples locally

All examples connect to LocalStack. They auto-create the stream on first run.

# Start LocalStack (if not already running):
docker run --rm -p 4566:4566 localstack/localstack:3.3.0

# Then in separate terminals:
task example:producer                          # send 100 records
task example:consumer                          # consume them
task example:efo                               # EFO consumer
WORKER_ID=w-1 task example:multi-worker       # worker 1 of 3

Debugging in VS Code

Launch configurations are in .vscode/launch.json. Open the Run and Debug panel (⇧⌘D) and choose:

Config	What it does
`Example: basic-consumer (LocalStack)`	Debug the polling consumer against LocalStack on `:4566`
`Example: producer (LocalStack)`	Debug the producer
`Tests: current file`	Debug tests in the open file. Select a function name first to scope to one test.
`Tests: current file (integration tag)`	Same, but compiles with `-tags=integration` and sets `INTEGRATION=1`
`Integration: E2E tests`	Debug the full E2E suite (starts LocalStack via testcontainers)

Tip: Set breakpoints, then hit F5. The race detector is enabled via GOFLAGS=-race in all test configs.

Adding a test

Unit test — add _test.go next to the package under test. No build tag needed.
Integration test — add to test/integration/, start the file with //go:build integration, guard with if os.Getenv("INTEGRATION") != "1" { t.Skip(...) }, and use helpers.StartLocalStack(t) for AWS infra.

Common tasks

task lint          # run golangci-lint
task lint:fix      # lint + auto-fix
task vet           # go vet
task gen           # regenerate mocks (go generate ./...)
task tidy          # go mod tidy
task ci            # lint + unit tests (same as CI)
task ci:full       # lint + unit + integration tests

Architecture Diagrams

Producer — record flow

sequenceDiagram
    participant App as Application
    participant P as Producer
    participant Buf as Internal Buffer<br/>(chan ProducerRecord)
    participant SendLoop as sendLoop<br/>(goroutine)
    participant K as Kinesis<br/>PutRecords API

    App->>P: Put(ctx, pk, data)
    P->>Buf: enqueue record
    Note over Buf,SendLoop: sendLoop drains buffer on BatchSize,<br/>BatchBytes limit, or 100 ms ticker
    SendLoop->>SendLoop: accumulate batch
    SendLoop->>K: PutRecords (≤500 records, ≤5 MB)
    K-->>SendLoop: results (per-record success/fail)
    SendLoop->>SendLoop: retry failed records (C-018)
    App->>P: Flush(ctx)
    P->>SendLoop: signal flush via flushCh
    SendLoop->>SendLoop: flush internal batch
    SendLoop-->>P: ack (resume chan closed)
    P->>Buf: drainBuffer exclusively
    P-->>App: return nil
    App->>P: Close(ctx)
    P->>SendLoop: close stopCh
    SendLoop-->>P: wg.Done()
    P->>Buf: final drainBuffer
    P-->>App: return nil

Consumer — polling mode (single worker)

sequenceDiagram
    participant App as Application
    participant C as Consumer
    participant Coord as Coordinator
    participant LeaseMgr as LeaseManager<br/>(DynamoDB / Postgres)
    participant Worker as PollingWorker<br/>(per shard goroutine)
    participant Proc as RecordProcessor<br/>(your code)
    participant K as Kinesis<br/>GetRecords API

    App->>C: Start(ctx)
    C->>Coord: run(ctx)
    loop every ShardSyncInterval
        Coord->>K: ListShards
        K-->>Coord: open shards
        Coord->>LeaseMgr: CreateLeaseIfNotExists(shardID)
        Coord->>LeaseMgr: ListLeases
        LeaseMgr-->>Coord: all leases + owners
        Coord->>Coord: targetLeaseCount = ⌈shards/workers⌉
        Coord->>LeaseMgr: TakeLease / StealLease (expired)
        Coord->>Worker: startWorker(shardID) [goroutine]
    end
    Worker->>Proc: Initialize(shardID, checkpoint)
    loop poll loop
        Worker->>K: GetShardIterator / GetRecords
        K-->>Worker: records + nextIterator
        Worker->>Proc: ProcessRecords(records, checkpointer, lag)
        Proc->>LeaseMgr: Checkpoint(seqNum)
        Worker->>LeaseMgr: RenewLease (heartbeat)
    end
    App->>C: Stop()
    C->>Coord: cancel context
    Worker->>Proc: ShutdownRequested / ShardEnded / LeaseLost
    Worker-->>Coord: goroutine exits
    Coord-->>C: errgroup.Wait()
    C-->>App: return

Consumer — Enhanced Fan-Out (EFO) mode

sequenceDiagram
    participant Coord as Coordinator
    participant Sub as EFO Subscriber
    participant Worker as EFOWorker<br/>(per shard goroutine)
    participant K as Kinesis<br/>SubscribeToShard API
    participant Proc as RecordProcessor

    Coord->>Sub: Register(ctx)
    Sub->>K: RegisterStreamConsumer
    K-->>Sub: consumerARN (CREATING)
    Sub->>K: DescribeStreamConsumer (poll)
    K-->>Sub: status = ACTIVE
    Coord->>Worker: startWorker(shardID)
    loop reconnect loop (exponential backoff)
        Worker->>Sub: Subscribe(shardID, startSeq)
        Sub->>K: SubscribeToShard (HTTP/2)
        K-->>Sub: event stream (push)
        loop stream events
            K->>Worker: SubscribeToShardEvent{Records, ContinuationSeq}
            Worker->>Proc: ProcessRecords(records, checkpointer, lag)
            Proc->>Worker: Checkpoint(seqNum)
        end
        Note over Worker,K: Stream closed by AWS (max 5 min)<br/>or on error — reconnect with backoff
    end

Distributed lease coordination — multi-worker

stateDiagram-v2
    direction LR

    state "Lease: UNCLAIMED" as unclaimed
    state "Lease: OWNED by Worker-A" as ownedA
    state "Lease: OWNED by Worker-B" as ownedB
    state "Lease: EXPIRED" as expired

    [*] --> unclaimed: coordinator creates<br/>leaseRecord on first sync

    unclaimed --> ownedA: Worker-A calls TakeLease<br/>(conditional PutItem)
    unclaimed --> ownedB: Worker-B calls TakeLease<br/>(race — only one wins)

    ownedA --> ownedA: Worker-A increments<br/>leaseCounter every HeartbeatInterval
    ownedA --> expired: Worker-A crashes —<br/>counter stops incrementing
    expired --> ownedB: Worker-B detects expiry<br/>(counter unchanged since last sync)<br/>calls StealLease

    ownedB --> unclaimed: Worker-B calls ReleaseLease<br/>on graceful shutdown
    ownedA --> unclaimed: Worker-A calls ReleaseLease<br/>on graceful shutdown

Shard assignment across workers

graph LR
    subgraph "Kinesis Stream (4 shards)"
        S0[Shard 0]
        S1[Shard 1]
        S2[Shard 2]
        S3[Shard 3]
    end

    subgraph "DynamoDB Lease Table"
        L0["leaseKey=Shard-0<br/>owner=Worker-1<br/>counter=42"]
        L1["leaseKey=Shard-1<br/>owner=Worker-1<br/>counter=38"]
        L2["leaseKey=Shard-2<br/>owner=Worker-2<br/>counter=51"]
        L3["leaseKey=Shard-3<br/>owner=Worker-3<br/>counter=29"]
    end

    subgraph "Worker-1 (Process A)"
        W1P0[PollingWorker<br/>Shard-0]
        W1P1[PollingWorker<br/>Shard-1]
    end

    subgraph "Worker-2 (Process B)"
        W2P2[PollingWorker<br/>Shard-2]
    end

    subgraph "Worker-3 (Process C)"
        W3P3[PollingWorker<br/>Shard-3]
    end

    S0 --> W1P0
    S1 --> W1P1
    S2 --> W2P2
    S3 --> W3P3

    W1P0 -.->|heartbeat| L0
    W1P1 -.->|heartbeat| L1
    W2P2 -.->|heartbeat| L2
    W3P3 -.->|heartbeat| L3

Checkpoint flow and shard lifecycle

stateDiagram-v2
    direction TB

    [*] --> Initialize: worker acquires lease

    Initialize --> ProcessRecords: Initialize(shardID, lastCheckpoint)

    ProcessRecords --> ProcessRecords: batch delivered\nCheckpoint(seqNum) written\nto DynamoDB

    ProcessRecords --> ShutdownRequested: Stop() called\n(graceful drain)
    ProcessRecords --> LeaseLost: another worker\nstole the lease
    ProcessRecords --> ShardEnded: shard closed\n(parent of reshard)

    ShutdownRequested --> [*]: checkpoint preserved\nnew worker resumes here
    LeaseLost --> [*]: no checkpoint\nnew owner re-reads from last checkpoint
    ShardEnded --> [*]: CheckpointTrimHorizon()\nchild shards can start

    note right of ShardEnded
        Child shards only start
        after parent reaches
        TRIM_HORIZON checkpoint
    end note

Internal package structure

graph TD
    subgraph "Public API  (pkg/)"
        consumer["pkg/consumer\nConsumer interface\nCoordinator\nOptions"]
        producer["pkg/producer\nProducer interface\nOptions"]
        middleware["pkg/middleware\nChain()"]
        logger["pkg/logger\nLogger interface\nSlogAdapter / Noop"]
        metrics["pkg/metrics\nMetricsClient interface\nNoopMetricsClient"]
        checkpoint["pkg/checkpoint\nCheckpointer interface"]
        lease["pkg/lease\nStore interface\nLease struct"]
        errors["pkg/errors\nSentinel errors\nIsRetryable()"]
        aggregation["pkg/aggregation\nDeaggregateRecords()"]
    end

    subgraph "Internal  (internal/)"
        intworker["internal/worker\nPollingWorker\nEFOWorker"]
        intcoord["internal/lease\nManager\nDynamoDBStore\nPostgresStore"]
        intefo["internal/efo\nSubscriber\nSDKEventStreamAdapter"]
        intshard["internal/shard\nDiscoverer"]
        intcheckpoint["internal/checkpoint\nCheckpointer\n(lease-backed, any backend)"]
        intretry["internal/retry\nDo() with backoff"]
        intmetrics["internal/metrics\nVictoriaMetrics\nCloudWatch\nMultiClient"]
    end

    consumer --> intworker
    consumer --> intcoord
    consumer --> intshard
    consumer --> intefo
    consumer --> intcheckpoint
    producer --> intretry
    intworker --> intefo
    intworker --> intretry
    intworker --> checkpoint
    intcoord --> lease
    intcheckpoint --> checkpoint
    consumer --> middleware
    consumer --> logger
    consumer --> metrics
    producer --> logger
    producer --> metrics

Examples

Example	Command
Basic polling consumer	`LOCALSTACK=1 go run ./examples/basic-consumer`
Enhanced Fan-Out	`LOCALSTACK=1 go run ./examples/efo-consumer`
Multi-worker distributed	`LOCALSTACK=1 WORKER_ID=w-1 go run ./examples/multi-worker`
Producer	`LOCALSTACK=1 go run ./examples/producer`

Name		Name	Last commit message	Last commit date
Latest commit History 5 Commits
.github/workflows		.github/workflows
.vscode		.vscode
examples		examples
internal		internal
pkg		pkg
test		test
.gitignore		.gitignore
LICENSE		LICENSE
README.md		README.md
Taskfile.yml		Taskfile.yml
go.mod		go.mod
go.sum		go.sum

Folders and files

Latest commit

History

Repository files navigation