Change Data Capture with Debezium and Go
Building production-ready real-time data pipelines with Debezium CDC and Go. Performance optimization, failure handling, and production lessons learned.
Key Takeaways
- Real-time sync: Process database changes within milliseconds
- Scalability: Handle millions of events with minimal latency
- Reliability: Zero data loss with exactly-once semantics
- Simplicity: No database polling or complex ETL jobs
The Problem: Real-Time Data Without Complexity
At my last company, we had 47 microservices all trying to stay in sync. The typical approach was REST APIs everywhere, database polling, and prayer.
Then we discovered Change Data Capture (CDC). Instead of asking "what changed?", CDC tells you immediately when something changes in your database.
Setting Up Debezium with PostgreSQL
Debezium is a distributed platform for change data capture. It monitors your databases and streams all changes to Kafka.
# docker-compose.yml for development
version: '3.8'
services:
zookeeper:
image: confluentinc/cp-zookeeper:7.4.0
environment:
ZOOKEEPER_CLIENT_PORT: 2181
kafka:
image: confluentinc/cp-kafka:7.4.0
depends_on: [zookeeper]
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
postgres:
image: postgres:15
environment:
POSTGRES_DB: testdb
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
command: postgres -c wal_level=logical
connect:
image: debezium/connect:2.4
depends_on: [kafka, postgres]
environment:
BOOTSTRAP_SERVERS: kafka:9092
GROUP_ID: debezium
CONFIG_STORAGE_TOPIC: connect_configs
OFFSET_STORAGE_TOPIC: connect_offsets
STATUS_STORAGE_TOPIC: connect_statusesThe key here is wal_level=logical for PostgreSQL. This enables logical decoding, which Debezium uses to capture changes.
Go Consumer Implementation
Here's a production-ready Go consumer that can handle millions of events per day:
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"os"
"os/signal"
"sync"
"syscall"
"time"
"github.com/IBM/sarama"
)
type CDCEvent struct {
Schema struct {
Name string `json:"name"`
Fields []struct {
Field string `json:"field"`
Type string `json:"type"`
} `json:"fields"`
} `json:"schema"`
Payload struct {
Before map[string]interface{} `json:"before"`
After map[string]interface{} `json:"after"`
Source struct {
Version string `json:"version"`
Connector string `json:"connector"`
Name string `json:"name"`
TsMs int64 `json:"ts_ms"`
Snapshot string `json:"snapshot"`
Db string `json:"db"`
Sequence string `json:"sequence"`
Schema string `json:"schema"`
Table string `json:"table"`
TxId int64 `json:"txId"`
Lsn int64 `json:"lsn"`
Xmin int64 `json:"xmin"`
} `json:"source"`
Op string `json:"op"` // c=create, u=update, d=delete, r=read
TsMs int64 `json:"ts_ms"`
TransactionData struct {
Id string `json:"id"`
TotalOrder int64 `json:"total_order"`
DataCollectionOrder int64 `json:"data_collection_order"`
} `json:"transaction"`
} `json:"payload"`
}
type EventProcessor struct {
handlers map[string]func(CDCEvent) error
metrics *Metrics
}
type Metrics struct {
EventsProcessed int64
ProcessingTime time.Duration
ErrorCount int64
LastProcessedTime time.Time
mu sync.RWMutex
}
func NewEventProcessor() *EventProcessor {
return &EventProcessor{
handlers: make(map[string]func(CDCEvent) error),
metrics: &Metrics{},
}
}
func (ep *EventProcessor) RegisterHandler(table string, handler func(CDCEvent) error) {
ep.handlers[table] = handler
}
func (ep *EventProcessor) ProcessEvent(event CDCEvent) error {
start := time.Now()
defer func() {
ep.metrics.mu.Lock()
ep.metrics.EventsProcessed++
ep.metrics.ProcessingTime += time.Since(start)
ep.metrics.LastProcessedTime = time.Now()
ep.metrics.mu.Unlock()
}()
table := event.Payload.Source.Table
handler, exists := ep.handlers[table]
if !exists {
return fmt.Errorf("no handler for table %s", table)
}
if err := handler(event); err != nil {
ep.metrics.mu.Lock()
ep.metrics.ErrorCount++
ep.metrics.mu.Unlock()
return fmt.Errorf("handler error for table %s: %w", table, err)
}
return nil
}
type Consumer struct {
processor *EventProcessor
consumer sarama.ConsumerGroup
}
func (c *Consumer) Setup(sarama.ConsumerGroupSession) error { return nil }
func (c *Consumer) Cleanup(sarama.ConsumerGroupSession) error { return nil }
func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for {
select {
case message := <-claim.Messages():
if message == nil {
return nil
}
var event CDCEvent
if err := json.Unmarshal(message.Value, &event); err != nil {
log.Printf("Failed to unmarshal message: %v", err)
session.MarkMessage(message, "")
continue
}
if err := c.processor.ProcessEvent(event); err != nil {
log.Printf("Failed to process event: %v", err)
// Don't mark as processed on error - this allows for retry
continue
}
session.MarkMessage(message, "")
case <-session.Context().Done():
return nil
}
}
}
func main() {
config := sarama.NewConfig()
config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
config.Consumer.Offsets.Initial = sarama.OffsetNewest
config.Consumer.Group.Session.Timeout = 10 * time.Second
config.Consumer.Group.Heartbeat.Interval = 3 * time.Second
brokers := []string{"localhost:9092"}
topics := []string{"testserver.public.users", "testserver.public.orders"}
groupID := "cdc-processor"
consumer, err := sarama.NewConsumerGroup(brokers, groupID, config)
if err != nil {
log.Fatal("Failed to create consumer group:", err)
}
processor := NewEventProcessor()
// Register handlers for different tables
processor.RegisterHandler("users", handleUserChanges)
processor.RegisterHandler("orders", handleOrderChanges)
cdcConsumer := &Consumer{
processor: processor,
consumer: consumer,
}
ctx, cancel := context.WithCancel(context.Background())
// Start metrics reporter
go reportMetrics(processor.metrics, 30*time.Second)
// Start consumer
go func() {
for {
if err := consumer.Consume(ctx, topics, cdcConsumer); err != nil {
log.Printf("Error from consumer: %v", err)
time.Sleep(1 * time.Second)
}
if ctx.Err() != nil {
return
}
}
}()
// Wait for interrupt signal
sigterm := make(chan os.Signal, 1)
signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
<-sigterm
cancel()
consumer.Close()
}
func handleUserChanges(event CDCEvent) error {
switch event.Payload.Op {
case "c": // create
log.Printf("User created: %+v", event.Payload.After)
// Update search index, send welcome email, etc.
case "u": // update
log.Printf("User updated: %+v", event.Payload.After)
// Invalidate cache, update derived data, etc.
case "d": // delete
log.Printf("User deleted: %+v", event.Payload.Before)
// Cleanup related data, audit log, etc.
}
return nil
}
func handleOrderChanges(event CDCEvent) error {
switch event.Payload.Op {
case "c":
log.Printf("Order created: %+v", event.Payload.After)
// Send confirmation email, update inventory, etc.
case "u":
log.Printf("Order updated: %+v", event.Payload.After)
// Update fulfillment system, recalculate metrics, etc.
}
return nil
}
func reportMetrics(metrics *Metrics, interval time.Duration) {
ticker := time.NewTicker(interval)
defer ticker.Stop()
for range ticker.C {
metrics.mu.RLock()
log.Printf("Metrics: processed=%d, errors=%d, avg_time=%v, last_processed=%v",
metrics.EventsProcessed,
metrics.ErrorCount,
metrics.ProcessingTime/time.Duration(max(metrics.EventsProcessed, 1)),
metrics.LastProcessedTime.Format(time.RFC3339))
metrics.mu.RUnlock()
}
}
func max(a, b int64) int64 {
if a > b {
return a
}
return b
}Performance Optimization Techniques
1. Batch Processing
Instead of processing events one by one, batch them for better throughput:
type BatchProcessor struct {
events []CDCEvent
batchSize int
flushTime time.Duration
processor func([]CDCEvent) error
}
func (bp *BatchProcessor) Add(event CDCEvent) error {
bp.events = append(bp.events, event)
if len(bp.events) >= bp.batchSize {
return bp.flush()
}
return nil
}
func (bp *BatchProcessor) flush() error {
if len(bp.events) == 0 {
return nil
}
if err := bp.processor(bp.events); err != nil {
return err
}
bp.events = bp.events[:0] // Reset slice but keep capacity
return nil
}2. Event Filtering
Filter events early to reduce processing overhead:
func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
for message := range claim.Messages() {
// Quick check before unmarshaling
if !c.shouldProcess(message.Topic, message.Key) {
session.MarkMessage(message, "")
continue
}
var event CDCEvent
if err := json.Unmarshal(message.Value, &event); err != nil {
log.Printf("Failed to unmarshal: %v", err)
session.MarkMessage(message, "")
continue
}
// More detailed filtering
if c.shouldSkipEvent(event) {
session.MarkMessage(message, "")
continue
}
if err := c.processor.ProcessEvent(event); err != nil {
log.Printf("Processing error: %v", err)
continue
}
session.MarkMessage(message, "")
}
return nil
}Handling Schema Evolution
Database schemas change. Your CDC consumer needs to handle this gracefully:
type SchemaRegistry struct {
schemas map[string]Schema
mu sync.RWMutex
}
type Schema struct {
Version int
Fields map[string]FieldInfo
CreatedAt time.Time
}
type FieldInfo struct {
Type string
Optional bool
Default interface{}
}
func (sr *SchemaRegistry) GetField(schemaName, fieldName string, event CDCEvent) (interface{}, bool) {
sr.mu.RLock()
schema, exists := sr.schemas[schemaName]
sr.mu.RUnlock()
if !exists {
// Register new schema
sr.registerSchema(schemaName, event.Schema)
return sr.GetField(schemaName, fieldName, event)
}
value, exists := event.Payload.After[fieldName]
if !exists {
// Check if field has default value
if fieldInfo, ok := schema.Fields[fieldName]; ok && fieldInfo.Default != nil {
return fieldInfo.Default, true
}
return nil, false
}
return value, true
}Error Handling and Recovery
Production systems fail. Here's how to handle it gracefully:
type ErrorHandler struct {
deadLetterTopic string
producer sarama.SyncProducer
maxRetries int
}
func (eh *ErrorHandler) HandleError(event CDCEvent, err error, retryCount int) error {
if retryCount < eh.maxRetries {
// Exponential backoff
backoff := time.Duration(1<<retryCount) * time.Second
time.Sleep(backoff)
return fmt.Errorf("retry %d/%d: %w", retryCount+1, eh.maxRetries, err)
}
// Send to dead letter queue
return eh.sendToDeadLetter(event, err)
}
func (eh *ErrorHandler) sendToDeadLetter(event CDCEvent, originalErr error) error {
message := &sarama.ProducerMessage{
Topic: eh.deadLetterTopic,
Value: sarama.StringEncoder(fmt.Sprintf(
"Failed to process: %v\nOriginal event: %+v",
originalErr, event)),
Headers: []sarama.RecordHeader{
{Key: []byte("error"), Value: []byte(originalErr.Error())},
{Key: []byte("timestamp"), Value: []byte(time.Now().Format(time.RFC3339))},
},
}
_, _, err := eh.producer.SendMessage(message)
return err
}Production Lessons Learned
1. Monitor Lag
Consumer lag is your most important metric. If lag grows, you're falling behind.
func monitorLag(consumer sarama.ConsumerGroup, topics []string) {
ticker := time.NewTicker(30 * time.Second)
defer ticker.Stop()
client, _ := sarama.NewClient([]string{"localhost:9092"}, nil)
defer client.Close()
for range ticker.C {
coordinator, err := client.Coordinator("cdc-processor")
if err != nil {
continue
}
request := &sarama.OffsetFetchRequest{Version: 1, ConsumerGroup: "cdc-processor"}
for _, topic := range topics {
partitions, _ := client.Partitions(topic)
for _, partition := range partitions {
request.AddPartition(topic, partition)
}
}
response, err := coordinator.FetchOffset(request)
if err != nil {
continue
}
for topic, partitions := range response.Blocks {
for partition, block := range partitions {
if block.Err != sarama.ErrNoError {
continue
}
latest, err := client.GetOffset(topic, partition, sarama.OffsetNewest)
if err != nil {
continue
}
lag := latest - block.Offset
log.Printf("Lag for %s-%d: %d messages", topic, partition, lag)
if lag > 10000 {
log.Printf("HIGH LAG WARNING: %s-%d has %d messages behind", topic, partition, lag)
}
}
}
}
}2. Handle Duplicates
CDC can deliver duplicates. Make your handlers idempotent:
type IdempotencyManager struct {
processed map[string]time.Time
mu sync.RWMutex
ttl time.Duration
}
func (im *IdempotencyManager) IsProcessed(eventID string) bool {
im.mu.RLock()
defer im.mu.RUnlock()
processedAt, exists := im.processed[eventID]
if !exists {
return false
}
// Check if entry is stale
if time.Since(processedAt) > im.ttl {
go im.cleanup(eventID)
return false
}
return true
}
func (im *IdempotencyManager) MarkProcessed(eventID string) {
im.mu.Lock()
defer im.mu.Unlock()
im.processed[eventID] = time.Now()
}
func createEventID(event CDCEvent) string {
return fmt.Sprintf("%s:%d:%d",
event.Payload.Source.Name,
event.Payload.Source.TxId,
event.Payload.TransactionData.TotalOrder)
}Performance Numbers
Typical CDC pipelines with Debezium and Go can achieve:
- Throughput: Millions to billions of events per day depending on hardware
- Latency: Sub-second replication from database to consumers
- Scalability: Horizontal scaling through Kafka partitions and consumer groups
- Resource efficiency: Low memory footprint with proper batching and buffering
- Reliability: At-least-once delivery guarantees with proper offset management
Common Pitfalls
1. Not Handling Tombstone Records
Deletes can create tombstone records (null payload). Handle them explicitly.
2. Ignoring Transaction Boundaries
Related changes might arrive in different messages. Use transaction IDs to group them.
3. Memory Leaks in Long-Running Consumers
Always reset slices and close channels properly. Use memory profiling regularly.
Conclusion
CDC with Debezium and Go gives you real-time data streaming without the complexity of distributed transactions or eventual consistency problems. The key is proper error handling, monitoring, and making your consumers idempotent.
Start simple: one table, one consumer, basic error handling. Then scale up as you learn the patterns and understand your data flow.