Forem: Vlad Pistun

Schema Evolution in Kafka: How to Design Resilient Event Contracts in Go

Vlad Pistun — Mon, 28 Jul 2025 18:48:41 +0000

In distributed systems, change is inevitable - but breaking things doesn’t have to be. As your system evolves, so will the data exchanged between services. In Kafka-based architectures, that evolution must be handled carefully. Fail to do so, and you'll end up with incompatible consumers, corrupted data, and replay nightmares.

This article dives into schema evolution in Kafka, with a sharp focus on using Protobuf and Go to build resilient contracts. We'll cover versioning strategies, evolution-safe patterns, common pitfalls, and practical examples using kafka-go and a schema registry (Buf or Confluent).

Why Schema Evolution Matters

When your services communicate through Kafka topics, they rely on a shared contract - the event schema. But these schemas aren’t static. You’ll inevitably need to:

Add new fields (e.g., tax_code to an Order event)
Deprecate fields (e.g., replacing total_amount with tax_amonut + taxable_amount)
Change types (e.g., int → decimal)

Without a disciplined approach to schema evolution, these changes break consumers, especially if they're deployed out-of-sync or written in different languages.

Breaking change example:
You remove a field that a downstream Go service still expects. On replay, it panics, or worse, silently processes corrupted data.

Kafka Doesn’t Understand Your Schema

Kafka stores bytes - not meaning. That means schema evolution isn't Kafka’s problem to solve. It's yours.

To manage this, most production-grade systems use a Schema Registry:

Confluent Schema Registry (for Avro, Protobuf, JSON Schema)
Buf Schema Registry

These registries:

Store schema versions
Check for compatibility (forward, backward, full)
Help consumers safely deserialize bytes

Protobuf: The Go-Friendly Choice

For Go services, Protobuf offers strong typing, excellent performance, and good evolution support.

Why not JSON or Avro?

JSON: No schema enforcement. Easy to break without noticing.
Avro: Schema evolution support is solid, but Go support is weak and verbose.

Why Protobuf Works Well

Fields can be marked optional
Defaults are implicit
Fields have tags (field numbers), so renaming is safe
Decoding skips unknown fields by default

Designing for Evolution: Best Practices

✅ 1. Add Fields, Don’t Remove
Additive changes are safest. Make new fields optional, and give them safe defaults on the consumer side.

message OrderV2 {
  string id = 1;
  int32 amount = 2;
  optional int32 tax = 3; // added in v2
}

In Go:

type OrderV2 struct {
  Id     string
  Amount int32
  Tax    *int32 // optional, nil if not present
}

❌ 2. Don’t Reuse Field Numbers
In Protobuf, field numbers are sacred. Changing the type of an existing field or reusing an old number for a new field leads to unpredictable behavior.

✅ 3. Default Tolerant Consumers
Consumers should handle missing fields gracefully and never rely on a field always being present.

Forward vs Backward Compatibility

Type	Description	Use Case
Forward	New consumers can read old events	Upgrade consumers first
Backward	Old consumers can read new events	Upgrade producers first
Full	Both directions safe	Best for long-term contracts

Topic vs Schema Versioning

There are two major strategies:

1. Versioned Schema, Same Topic

Topic remains orders
Schemas evolve (v1, v2, v3...)
Consumers must be aware of schema versions or use registry-based decoding

2. Versioned Topics (orders.v1, orders.v2)

Clear boundary, simpler consumers
Harder to replay across versions
Сan lead to topic sprawl

Recommended: Use versioned schemas + a schema registry, unless you have a good reason to version topics.

Conclusion

Schema evolution in Kafka is a challenge - but not an unsolvable one. With Protobuf, a proper schema registry, and evolution-safe Go code, you can build contracts that withstand change and avoid production surprises.

Design schemas like APIs. Make them clear, compatible, and built to last.

👀 Furthermore: Exploring Buf Stream

Buf recently introduced Buf Stream, a high-performance streaming schema transport designed to simplify Protobuf communication across services. While not yet widely adopted in Kafka ecosystems, it shows promise for future use cases involving real-time schema-bound messaging - especially where tight Protobuf integration and compatibility checks are critical.

At the time of writing, we haven’t tested Buf Stream in production or in Kafka contexts, but it's worth keeping an eye on as the ecosystem evolves.

Event-Carried State Transfer for Tax Calculation in Fintech Real-Time Architecture with Kafka, Go, and PostgreSQL

Vlad Pistun — Sun, 13 Jul 2025 19:28:40 +0000

In modern Fintech systems, microservices are distributed and autonomous — but often need access to shared state in real time. A common challenge arises when one service, like a Tax Service, needs to act on data owned by another, such as a Payments Service.

This article shows how to implement Event-Carried State Transfer (ECST) to solve this problem cleanly and reliably, using Kafka for messaging, Go for services, and PostgreSQL for durability. We'll explore how to calculate taxes in real-time — without REST calls, shared databases, or fragile coordination.

Architecture Overview

We’re working with two services:

Payments Service

Owns the Order and Transaction domains.
Emits orders.created with full order details.
Emits transactions.created as financial events happen.

Tax Service

Listens to both orders.created and transactions.created.
Builds a durable local cache of orders using PostgreSQL.
On each transaction, it retrieves the related order and calculates tax.
Emits a tax.calculated event with the enriched data.

How Kafka + Go + PostgreSQL Work Together

Kafka streams events with full entity state (orders.created, transactions.created).

Go handles the concurrency and performance needs of both producers and consumers.

PostgreSQL provides a durable local cache, so the Tax Service can store what it sees — and make decisions offline, quickly, and deterministically.

This trio is battle-tested and scales well under high volume, audit-heavy environments.

What is Event-Carried State Transfer?

Event-Carried State Transfer (ECST) is a messaging pattern where a service emits not just event metadata or IDs, but the entire state of an entity. That way, any downstream service can cache and act on it independently.

Protobuf Models (via Buf)

message Order {
  string order_id = 1;
  string customer_id = 2;
  string country = 3;
  string currency = 4;
  int32 amount = 5;
  string product_type = 6;
  google.protobuf.Timestamp created_at = 7;
  google.protobuf.Timestamp updated_at = 8;
}

message Transaction {
  string transaction_id = 1;
  string order_id = 2;
  string customer_id = 3;
  int32  amount = 4;
  string status = 5;
  google.protobuf.Timestamp created_at = 6;
  google.protobuf.Timestamp updated_at = 7;
}

PostgreSQL Schema for ECST Order Cache

Instead of querying Payments for every transaction, the Tax Service builds its own ECST snapshot of the Orders it needs — one row per order, stored as JSON.

Why JSONB?

You can query it for debugging or analytics.
You don’t need to tightly couple DB schema to Protobuf definitions.
You can store the exact payload you saw on the Kafka topic.

CREATE TABLE order_events (
    id VARCHAR PRIMARY KEY,
    payload JSONB NOT NULL,
    order_created_at TIMESTAMP(6) WITHOUT TIME ZONE NOT NULL,
    created_at      TIMESTAMP(6) WITHOUT TIME ZONE NOT NULL DEFAULT now(),
    updated_at      TIMESTAMP(6) WITHOUT TIME ZONE NOT NULL DEFAULT now()
);

id: order ID
payload: full order data as JSON (Protobuf → JSON)
order_created_at: original event timestamp
created_at: when the row was inserted
updated_at: when it was last updated

Writing Orders to the Cache

When orders.created events arrive, the Tax Service writes them to Postgres:

func upsertOrderJSON(db *sql.DB, order Order) error {
  jsonData, err := json.Marshal(order)
  if err != nil {
     return fmt.Errorf("marshal: %w", err)
  }
  _, err := db.Exec(`
    INSERT INTO order_cache (id, payload, order_created_at)
    VALUES ($1, $2, to_timestamp($3))
    ON CONFLICT (id) DO UPDATE
    SET payload = EXCLUDED.payload,
        updated_at = now()
  `, order.OrderID, jsonData, order.CreatedAt)
  return err
}

Reading Orders and Calculating Tax

When a transactions.created event arrives, the service fetches the order and calculates tax:

func getOrderByID(db *sql.DB, id string) (Order, error) {
  row := db.QueryRow(`SELECT payload FROM order_events WHERE id = $1`, id)
  var raw []byte
  err := row.Scan(&raw)
  if err != nil {
     return Order{}, fmt.Errorf("row scan: %w", err)
  }
  var order Order
  err = json.Unmarshal(raw, &order)
  if err != nil {
     return Order{}, fmt.Errorf("unmarshal: %w", err)
  }
  return order, nil
}

func calculateTax(order Order, tx Transaction) TaxedTransaction {
  vat := map[string]float64{"DE": 0.19, "FR": 0.20}[order.Country]
  if order.ProductType == "crypto" {
    vat = 0.0
  }
  tax := order.Amount * vat

  return &TaxedTransaction{
    TransactionID: tx.TransactionID,
    OrderID:       order.OrderID,
    CustomerID:    order.CustomerID,
    BaseAmount:    order.Amount,
    TaxApplied:    tax,
    TotalAmount:   order.Amount + tax,
    Country:       order.Country,
    ProductType:   order.ProductType,
  }
}

Handling Out-of-Order Events with Retries

Kafka doesn't guarantee that orders.created will arrive before transactions.created. So the Tax Service must retry:

func getOrderWithRetries(db *sql.DB, orderID string, maxRetries int, delay time.Duration) (Order, error) {
  for i := 0; i < maxRetries; i++ {
    order, err := getOrderByID(db, orderID)
    if err != nil {
        if errors.Is(err, sql.ErrNoRows) {
           time.Sleep(delay)
           delay *= 2
           continue
        }
        return Order{}, fmt.Errorf("get order: %w", err)
    }

    return order, nil
  }

  return Order{}, fmt.Errorf("order not found after retries: %s", orderID)
}

If the order is still missing, the service can emit the failed transaction to a tax.failed topic for later processing.

Benefits of This Design

No RPCs: The Tax Service does not depend on Payments being up or reachable.
Durable State: PostgreSQL acts as a reliable local cache.
Fast & Reactive: All tax logic happens locally in response to events.
Failure-tolerant: Retry logic handles timing issues. DLQ covers gaps.
Auditable: All order data and tax calculations are traceable and replayable.

Common Misconceptions About ECST

Isn't this denormalized?
Yes — and that's the point. You denormalize state so consumers don’t have to coordinate or join at runtime.

What if the schema changes?
Use Protobuf + Buf Schema Registry with breaking-change enforcement. You evolve your schemas safely.

Won’t Postgres get too big?
Only if you never prune or partition. But for most transaction-based services, retaining ~3 year of cache is trivial. Use TTLs or archive pipelines if needed.

Conclusion

Event-Carried State Transfer is a powerful pattern that gives microservices autonomy without sacrificing correctness. In Fintech, where correctness, auditability, and latency all matter, ECST with Kafka and Postgres offers a clean solution.

With this architecture, your services are resilient, reactive, and easy to reason about — no shared database, no RPC chains, and no surprises.

Handling Failures in Kafka Consumers: Go Patterns That Work in Production

Vlad Pistun — Sat, 05 Jul 2025 10:03:19 +0000

Introduction

Apache Kafka has become the backbone of modern event-driven architectures, enabling high-throughput pipelines, real-time analytics, and loosely coupled microservices. But building resilient Kafka consumers-especially in Go, a language that emphasizes simplicity and low-level control-presents a unique set of challenges.

In production, failures aren’t edge cases-they’re daily realities. Network partitions, broker restarts, corrupted messages, serialization mismatches, and poisoned pills can bring down consumers if not handled properly. Without thoughtful design, you risk message loss, duplication, retry storms, or system-wide downtime.

This article dives deep into production-tested patterns for handling failures in Kafka consumers written in Go. Using the excellent kafka-go library, we’ll cover:

Retry logic with exponential backoff
Dead-letter queues for unrecoverable messages
Manual offset commits for at-least-once delivery
Supervision patterns for resilient runtime behavior
Schema validation to catch bad data before it crashes your logic

Whether you’re building real-time systems or maintaining a high-throughput message processor, these Go patterns will help you design Kafka consumers that can gracefully survive the messiness of production.

🔁 1. Retry Logic with Backoff (for transient errors)

Scenario: Your consumer hits a timeout when fetching messages or a temporary processing failure.

Pattern:
Use exponential backoff with jitter to avoid retry storms and give downstream systems time to recover. Mostly, I prefer https://pkg.go.dev/github.com/cenkalti/backoff/v4 for retries.

Example:


func retryWithBackoff(fn func() error) error {
    backoff := time.Millisecond * 100
    for attempts := 0; attempts < 3; attempts++ {
        err := fn()
        if err == nil {
            return nil
        }
        time.Sleep(backoff + time.Duration(rand.Intn(100))*time.Millisecond)
        backoff *= 2
    }
    return errors.New("max retry attempts reached")
}

Apply to message processing:


for {
    m, err := r.ReadMessage(ctx)
    if err != nil {
        log.Printf("read error: %v", err)
        continue
    }

    retryWithBackoff(func() error {
        return handleMessage(m)
    })
}

🪵 2. Dead Letter Queue (DLQ) for poison pills

Scenario: A malformed message keeps causing panics or errors every time it's retried.

Pattern: After a fixed number of retries, send the message to a separate DLQ topic for analysis.

Example:

func handleWithDLQ(m kafka.Message, producer *kafka.Writer) {
    err := retryWithBackoff(func() error {
        return processMessage(m)
    })

    if err != nil {
        log.Printf("failed to process message: %v", err)
        dlqMessage := kafka.Message{
            Key:   m.Key,
            Value: m.Value,
            Headers: []kafka.Header{
                {Key: "error", Value: []byte(err.Error())},
            },
        }
        _ = producer.WriteMessages(context.Background(), dlqMessage)
    }
}

🔄 3. At-Least-Once Delivery with Manual Commit

Scenario: You want to ensure messages are never lost, even if the process crashes.

Pattern: Read messages, process them, and commit offsets after successful handling.

Example:

r := kafka.NewReader(kafka.ReaderConfig{
    Brokers:  []string{"localhost:9092"},
    Topic:    "events",
    GroupID:  "event-handler",
    MinBytes: 10e3,
    MaxBytes: 10e6,
    CommitInterval: 0, // disable auto-commit
})

for {
    m, err := r.ReadMessage(ctx)
    if err != nil {
        log.Printf("read error: %v", err)
        continue
    }

    if err := handleMessage(m); err == nil {
        _ = r.CommitMessages(ctx, m)
    } else {
        log.Printf("processing error: %v", err)
        // optionally write to DLQ
    }
}

🚦 4. Consumer Supervision & Graceful Restart

Scenario: Your consumer crashes due to an unhandled panic or dependency failure.

Pattern: Use a supervisor process that restarts consumers with metrics, backoff, and alerting.

Example (basic concept):

run := func() (err error) {
        defer func() {
            if r := recover(); r != nil {
                err = errors.Join(err, fmt.Errorf("panic: %v", r))
            }
        }()

        if err = runConsumer(); err != nil {
            return fmt.Errorf("run consumer: %w", err)
        }

        return
    }

    for {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }

        err := run()
        if err != nil && !errors.Is(err, context.Canceled) {
            logger.Error("run consumer", "err", err)
            time.Sleep(time.Second)
            continue
        }

        if err == nil {
            return nil
        }

        if ctx.Err() != nil {
            logger.Warn("consumer graceful shutdown timeout")
            return ctx.Err()
        }

        return nil
    }

🧪 5. Contract Testing with Schema Validation

Scenario: A consumer crashes due to unexpected payload shape.

Pattern: Validate against schema (Protobuf/Avro) at read-time.

With Go + Protobuf + Schema Registry:

Use Buf for linted schema definitions
Validate decoded messages before processing

msg := &MyEvent{}
if err := proto.Unmarshal(m.Value, msg); err != nil {
    log.Printf("unmarshal failed: %v", err)
    // send to DLQ
}

Or event u can use buf.build/go/protovalidate for validation proto

msg := &MyEvent{}
if err := proto.Unmarshal(m.Value, msg); err != nil {
    log.Printf("unmarshal failed: %v", err)
    // send to DLQ
}
err := protovalidate.Validate(msg)
if err != nil {
    log.Printf("validation failed: %v", err)
    // send to DLQ
}

✅ Final Thoughts

Failure handling in Kafka consumers is a balancing act between reliability, throughput, and cost. Go, with its simplicity and kafka-go's flexibility, gives you the right primitives to implement production-grade systems.

Checklist for resilient consumers:

Backoff and retries
Dead-letter routing
Manual offset commit
Panic recovery and supervision
Schema validation and message contracts