README
¶
outburst
A transactional outbox relay: it moves rows from a Postgres table to Kafka.
Usage
Import it as a library and start it during application boot:
var db *sqlx.DB = connectToDB()
err := outburst.Initialize(ctx, outburst.Options{
Kafka: kafkaProducer,
Database: db,
OutboxTable: "outbox",
})
Initialize provisions the table when it is missing and launches a background
relay: a LISTEN/NOTIFY consumer for low-latency delivery, backed by a
periodic sweeper that catches anything a missed notification left behind.
Notifying the relay
Have your insert trigger notify the kafka-message channel so a freshly written
row is forwarded right away; without a notification the sweeper still picks it up
on its next pass, just later.
The payload is a JSON object carrying the row id together with its kafka_key:
SELECT pg_notify('kafka-message', json_build_object('id', id, 'key', kafka_key)::text);
Shipping the key inside the notification lets the relay choose a worker shard
without a second query. Shards are keyed on kafka_key, so every row for a given
key is published in insertion order — and therefore lands on its Kafka partition
in order — regardless of WorkerCount.
Only this JSON shape is accepted. Any other payload is ignored on the notify path and left for the sweeper to forward.
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Initialize ¶
Initialize provisions the outbox table when needed and launches the background relay: a LISTEN/NOTIFY consumer plus periodic maintenance jobs. It returns once everything is wired; the relay keeps running until ctx is cancelled.
Types ¶
type Message ¶
type Message struct {
ID int64 `db:"id"`
Timestamp time.Time `db:"create_time"`
Topic string `db:"kafka_topic"`
Key sql.NullString `db:"kafka_key"`
Value []byte `db:"kafka_value"`
HeaderKeys ql.StringArray `db:"kafka_header_keys"`
HeaderValues ql.StringArray `db:"kafka_header_values"`
}
Message is a single outbox row ready to be published to Kafka.
type Options ¶
type Options struct {
// Producer used to publish outbox rows to Kafka.
Kafka *kafka.Producer
// Handle to the database that holds the outbox table.
Database *sqlx.DB
// Name of the outbox table to drain. Created on startup when it is missing.
OutboxTable string
// Scheduler that the periodic jobs attach to. When nil, outburst builds and
// owns its own scheduler.
Cron gocron.Scheduler
// Rows fetched per pass by the fallback sweeper. Defaults to 128.
BatchSize uint
// Size of the worker pool on the LISTEN/NOTIFY path. Defaults to 4.
//
// Every row is dispatched to shard hash(kafka_key)%WorkerCount and each
// shard publishes strictly in sequence. Two rows sharing a kafka_key thus
// reach Kafka in insertion order — and land on the same partition in that
// order — no matter how many workers run. Rows without a key carry no
// ordering guarantee and all fall to shard 0.
WorkerCount uint
// Capacity of each shard's hand-off channel on the NOTIFY path. Defaults to
// 128. Once a shard channel is full the LISTEN loop stops pulling new
// notifications until that shard drains. Grow it to ride out spikes, shrink
// it to bound the memory held in flight.
WorkerQueueBuffer uint
// Turn on verbose debug logging.
EnableDebugLogging bool
// contains filtered or unexported fields
}
Options configures the outburst outbox relay.
Source Files