Event-Driven Architecture with Go and Kafka

Introduction

As systems grow in complexity and scale, traditional request-response architectures often struggle to meet demands for scalability, resilience, and flexibility. Event-driven architecture (EDA) has emerged as a powerful pattern to address these challenges, enabling systems to react to changes in state through the production, detection, and consumption of events.

Over the past year, I've been designing and implementing event-driven systems using Go and Apache Kafka for several high-throughput applications. In this article, I'll share practical insights on building event-driven architectures with Go, covering event sourcing concepts, reliable message processing, handling schema evolution, and implementing exactly-once semantics.

Understanding Event-Driven Architecture

Before diving into the implementation details, let's clarify some key concepts in event-driven architecture:

Events vs. Messages

• Events represent facts that have occurred in the system. They're immutable records of something that happened at a specific point in time.
• Messages are the containers that transport events between services.

Event Sourcing

Event sourcing is a pattern where state changes are captured as a sequence of immutable events. Instead of storing just the current state, you store the full history of events, which can be replayed to reconstruct the state at any point in time.

Command Query Responsibility Segregation (CQRS)

CQRS separates read and write operations into different models. Commands (writes) update the event store, while queries (reads) use optimized read models derived from the events.

Benefits of Event-Driven Architecture

1. Loose Coupling: Services don't need direct knowledge of each other
2. Scalability: Components can scale independently
3. Resilience: Failure in one component doesn't bring down the entire system
4. Flexibility: Easier to adapt and extend the system
5. Auditability: Complete history of all state changes

Apache Kafka as an Event Backbone

Apache Kafka is a distributed streaming platform particularly well-suited for event-driven architectures:

Key Kafka Concepts

• Topics: Categories or feeds to which events are published
• Partitions: Divisions of topics for parallel processing
• Producers: Applications that publish events to topics
• Consumers: Applications that subscribe to topics and process events
• Consumer Groups: Groups of consumers that divide work among themselves
• Offsets: Pointers that track the last consumed event for each partition

Setting Up a Go Project with Kafka

Let's start by setting up a basic Go project structure for an event-driven application:

/event-driven-app
  /cmd
    /producer
      main.go
    /consumer
      main.go
  /internal
    /events
      events.go
      serialization.go
    /models
      user.go
      order.go
    /handlers
      user_handler.go
      order_handler.go
    /store
      event_store.go
      repository.go
  /pkg
    /kafka
      producer.go
      consumer.go
  go.mod
  go.sum

Basic Kafka Setup in Go

The github.com/segmentio/kafka-go library provides a clean, idiomatic Go client for Kafka:

// pkg/kafka/producer.go package kafka

import ( "context" "time"

"github.com/segmentio/kafka-go"

)

type Producer struct { writer *kafka.Writer }

func NewProducer(brokers []string) *Producer { writer := kafka.NewWriter(kafka.WriterConfig{ Brokers: brokers, BatchTimeout: 100 * time.Millisecond, Async: false, })

return &Producer{
    writer: writer,
}

}

func (p *Producer) Publish(ctx context.Context, topic string, key, value []byte) error { return p.writer.WriteMessages(ctx, kafka.Message{ Topic: topic, Key: key, Value: value, }) }

func (p *Producer) Close() error { return p.writer.Close() }

// pkg/kafka/consumer.go package kafka

import ( "context"

"github.com/segmentio/kafka-go"

)

type Consumer struct { reader *kafka.Reader }

func NewConsumer(brokers []string, topic, groupID string) *Consumer { reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: brokers, Topic: topic, GroupID: groupID, MinBytes: 10e3, // 10KB MaxBytes: 10e6, // 10MB })

return &Consumer{
    reader: reader,
}

}

func (c *Consumer) Read(ctx context.Context) (kafka.Message, error) { return c.reader.ReadMessage(ctx) }

func (c *Consumer) Close() error { return c.reader.Close() }

Event Design and Serialization

Properly designing your events is crucial for a successful event-driven system.

Event Structure

A well-designed event typically includes:

1. Event ID: Unique identifier for the event
2. Event Type: Describes what happened
3. Entity Type: The type of entity affected
4. Entity ID: The identifier of the affected entity
5. Timestamp: When the event occurred
6. Version: Schema version for evolution
7. Payload: The actual event data
8. Metadata: Additional contextual information

Here's an example in Go:

// internal/events/events.go package events

import ( "time"

"github.com/google/uuid"

)

type EventType string

const ( UserCreated EventType = "user.created" UserUpdated EventType = "user.updated" OrderPlaced EventType = "order.placed" OrderCompleted EventType = "order.completed" PaymentReceived EventType = "payment.received" )

type Event struct { ID string json:"id" Type EventType json:"type" EntityType string json:"entity_type" EntityID string json:"entity_id" Timestamp time.Time json:"timestamp" Version string json:"version" Payload interface{} json:"payload" Metadata map[string]string json:"metadata" }

func NewEvent(eventType EventType, entityType, entityID string, payload interface{}, metadata map[string]string) Event { if metadata == nil { metadata = make(map[string]string) }

return Event{
    ID:        uuid.New().String(),
    Type:      eventType,
    EntityType: entityType,
    EntityID:  entityID,
    Timestamp: time.Now().UTC(),
    Version:   "1.0",
    Payload:   payload,
    Metadata:  metadata,
}

}

Serialization Strategies

JSON is a common choice for event serialization due to its human-readability and flexibility, but it has performance and schema control drawbacks. Protocol Buffers or Avro are better choices for high-performance systems with strict schema control.

Let's implement JSON serialization for simplicity:

// internal/events/serialization.go package events

import ( "encoding/json" "fmt" )

func Serialize(event Event) ([]byte, error) { return json.Marshal(event) }

func Deserialize(data []byte) (Event, error) { var event Event if err := json.Unmarshal(data, &event); err != nil { return Event{}, fmt.Errorf("failed to deserialize event: %w", err) } return event, nil }

Implementing Event Sourcing

With event sourcing, all state changes are stored as a sequence of events. Here's a simple implementation:

// internal/store/event_store.go package store

import ( "context" "errors"

"github.com/yourusername/event-driven-app/internal/events"
"github.com/yourusername/event-driven-app/pkg/kafka"

)

var ( ErrEventNotFound = errors.New("event not found") )

type EventStore interface { Append(ctx context.Context, event events.Event) error Load(ctx context.Context, entityType, entityID string) ([]events.Event, error) }

type KafkaEventStore struct { producer *kafka.Producer consumer *kafka.Consumer topicPrefix string }

func NewKafkaEventStore(producer *kafka.Producer, consumer *kafka.Consumer, topicPrefix string) *KafkaEventStore { return &KafkaEventStore{ producer: producer, consumer: consumer, topicPrefix: topicPrefix, } }

func (s *KafkaEventStore) Append(ctx context.Context, event events.Event) error { // Serialize the event data, err := events.Serialize(event) if err != nil { return err }

// Derive topic name from entity type
topic := s.topicPrefix + event.EntityType

// Use entity ID as message key for partitioning
return s.producer.Publish(ctx, topic, []byte(event.EntityID), data)

}

func (s *KafkaEventStore) Load(ctx context.Context, entityType, entityID string) ([]events.Event, error) { // This is a simplified implementation // In a real system, you would use a database to store and query events // For simplicity, we're just showing the concept return nil, ErrEventNotFound }

Building Event Producers

Now, let's implement a simple event producer:

// cmd/producer/main.go package main

import ( "context" "encoding/json" "log" "time"

"github.com/yourusername/event-driven-app/internal/events"
"github.com/yourusername/event-driven-app/internal/models"
"github.com/yourusername/event-driven-app/internal/store"
"github.com/yourusername/event-driven-app/pkg/kafka"

)

func main() { // Initialize Kafka producer producer := kafka.NewProducer([]string{"localhost:9092"}) defer producer.Close()

// Initialize event store
eventStore := store.NewKafkaEventStore(producer, nil, "events.")

// Create a user
user := models.User{
    ID:        "user-123",
    Email:     "user@example.com",
    FirstName: "John",
    LastName:  "Doe",
    CreatedAt: time.Now().UTC(),
}

// Create user event
userCreatedEvent := events.NewEvent(
    events.UserCreated,
    "user",
    user.ID,
    user,
    map[string]string{
        "source": "user-service",
    },
)

// Publish the event
ctx := context.Background()
if err := eventStore.Append(ctx, userCreatedEvent); err != nil {
    log.Fatalf("Failed to publish event: %v", err)
}

log.Printf("Published event: %s", userCreatedEvent.ID)

}

Implementing Event Consumers

Let's implement a consumer that processes user events:

// cmd/consumer/main.go package main

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

"github.com/yourusername/event-driven-app/internal/events"
"github.com/yourusername/event-driven-app/internal/handlers"
"github.com/yourusername/event-driven-app/pkg/kafka"

)

func main() { // Initialize Kafka consumer consumer := kafka.NewConsumer( []string{"localhost:9092"}, "events.user", "user-processor", ) defer consumer.Close()

// Initialize handlers
userHandler := handlers.NewUserHandler()

// Create a context that cancels on SIGTERM/SIGINT
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// Handle OS signals
sigChan := make(chan os.Signal, 1)
signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
go func() {
    <-sigChan
    log.Println("Received termination signal, shutting down...")
    cancel()
}()

// Process events
log.Println("Starting consumer, waiting for events...")
for {
    select {
    case <-ctx.Done():
        log.Println("Context cancelled, exiting...")
        return
    default:
        // Get the next message
        msg, err := consumer.Read(ctx)
        if err != nil {
            if ctx.Err() != nil {
                // Context was cancelled, exit gracefully
                return
            }
            log.Printf("Error reading message: %v", err)
            continue
        }
        
        // Deserialize the event
        event, err := events.Deserialize(msg.Value)
        if err != nil {
            log.Printf("Error deserializing event: %v", err)
            continue
        }
        
        // Process the event based on its type
        log.Printf("Received event: %s of type %s", event.ID, event.Type)
        
        switch event.Type {
        case events.UserCreated:
            if err := userHandler.HandleUserCreated(ctx, event); err != nil {
                log.Printf("Error handling UserCreated event: %v", err)
            }
        case events.UserUpdated:
            if err := userHandler.HandleUserUpdated(ctx, event); err != nil {
                log.Printf("Error handling UserUpdated event: %v", err)
            }
        default:
            log.Printf("Unknown event type: %s", event.Type)
        }
    }
}

}

// internal/handlers/user_handler.go package handlers

import ( "context" "encoding/json" "fmt"

"github.com/yourusername/event-driven-app/internal/events"
"github.com/yourusername/event-driven-app/internal/models"

)

type UserHandler struct { // In a real application, this would have dependencies like repositories }

func NewUserHandler() *UserHandler { return &UserHandler{} }

func (h *UserHandler) HandleUserCreated(ctx context.Context, event events.Event) error { var user models.User

// Convert the payload to our user model
payloadJSON, err := json.Marshal(event.Payload)
if err != nil {
    return fmt.Errorf("failed to marshal payload: %w", err)
}

if err := json.Unmarshal(payloadJSON, &user); err != nil {
    return fmt.Errorf("failed to unmarshal user: %w", err)
}

// In a real application, you would save the user to a database
// For this example, we'll just log it
fmt.Printf("Processed UserCreated event: User %s (%s) created\n", user.ID, user.Email)

return nil

}

func (h *UserHandler) HandleUserUpdated(ctx context.Context, event events.Event) error { // Similar to HandleUserCreated return nil }

Handling Schema Evolution

As your system evolves, your event schemas will change. Here are some strategies for handling schema evolution:

Versioning Events

Include a version field in your events:

// Improved event struct with explicit versioning type Event struct { // ... other fields SchemaVersion string json:"schema_version" // ... other fields }

Backward and Forward Compatibility

• Backward compatibility: New consumers can process old events
• Forward compatibility: Old consumers can process new events

To maintain compatibility:

1. Always add fields as optional
2. Never remove fields (mark them as deprecated instead)
3. Never change field types
4. Use versioning to handle breaking changes

Schema Registry

For complex systems, consider using a schema registry like Confluent Schema Registry. While there's no official Go client, you can use third-party clients or the REST API directly.

Implementing Exactly-Once Processing

One of the challenges in event-driven systems is ensuring reliable message processing. There are three delivery semantics:

1. At-most-once: Messages might be lost but never processed twice
2. At-least-once: Messages are never lost but might be processed multiple times
3. Exactly-once: Messages are processed exactly once

Achieving exactly-once semantics requires combining at-least-once delivery with idempotent operations.

Idempotent Consumers

Make your event handlers idempotent so that processing the same event multiple times has the same effect as processing it once:

func (h *UserHandler) HandleUserCreated(ctx context.Context, event events.Event) error { var user models.User

// Deserialize user...

// Check if we've already processed this event
if h.eventRepository.HasProcessed(event.ID) {
    return nil // Already processed, skip
}

// Process the event
// ...

// Mark as processed
return h.eventRepository.MarkProcessed(event.ID)

}

Transactional Outbox Pattern

When you need to update a database and publish events atomically, use the Transactional Outbox pattern:

1. Within a database transaction:

   • Update the application state
   • Insert the event into an "outbox" table

2. A separate process reads from the outbox table and publishes events to Kafka

This ensures that the database update and event publication are atomic, even if the process crashes after the transaction commits but before publishing to Kafka.

Building a Complete Event-Driven System

Let's bring everything together with a more complete example of an order processing system:

Domain Models

// internal/models/order.go package models

import ( "time" )

type OrderStatus string

const ( OrderStatusPending OrderStatus = "pending" OrderStatusPaid OrderStatus = "paid" OrderStatusShipped OrderStatus = "shipped" OrderStatusDelivered OrderStatus = "delivered" OrderStatusCancelled OrderStatus = "cancelled" )

type OrderItem struct { ProductID string json:"product_id" Quantity int json:"quantity" Price float64 json:"price" }

type Order struct { ID string json:"id" UserID string json:"user_id" Status OrderStatus json:"status" Items []OrderItem json:"items" Total float64 json:"total" CreatedAt time.Time json:"created_at" UpdatedAt time.Time json:"updated_at" }

// Calculate total method func (o *Order) CalculateTotal() { var total float64 for _, item := range o.Items { total += item.Price * float64(item.Quantity) } o.Total = total }

Event Definitions

// internal/events/order_events.go package events

import ( "github.com/yourusername/event-driven-app/internal/models" )

// Event payloads type OrderPlacedPayload struct { Order models.Order json:"order" UserEmail string json:"user_email" }

type OrderPaidPayload struct { OrderID string json:"order_id" PaymentID string json:"payment_id" Amount float64 json:"amount" PaymentMethod string json:"payment_method" }

type OrderShippedPayload struct { OrderID string json:"order_id" TrackingNumber string json:"tracking_number" ShippedAt time.Time json:"shipped_at" Carrier string json:"carrier" }

// Helper functions to create specific events func NewOrderPlacedEvent(order models.Order, userEmail string) Event { payload := OrderPlacedPayload{ Order: order, UserEmail: userEmail, }

return NewEvent(
    OrderPlaced,
    "order",
    order.ID,
    payload,
    map[string]string{
        "source": "order-service",
    },
)

}

func NewOrderPaidEvent(orderID, paymentID string, amount float64, paymentMethod string) Event { payload := OrderPaidPayload{ OrderID: orderID, PaymentID: paymentID, Amount: amount, PaymentMethod: paymentMethod, }

return NewEvent(
    OrderPaid,
    "order",
    orderID,
    payload,
    map[string]string{
        "source": "payment-service",
    },
)

}

Command Handlers

// internal/commands/place_order.go package commands

import ( "context" "fmt" "time"

"github.com/google/uuid"
"github.com/yourusername/event-driven-app/internal/events"
"github.com/yourusername/event-driven-app/internal/models"
"github.com/yourusername/event-driven-app/internal/store"

)

type PlaceOrderCommand struct { UserID string json:"user_id" Items []models.OrderItem json:"items" UserEmail string json:"user_email" }

type PlaceOrderHandler struct { eventStore store.EventStore }

func NewPlaceOrderHandler(eventStore store.EventStore) *PlaceOrderHandler { return &PlaceOrderHandler{ eventStore: eventStore, } }

func (h *PlaceOrderHandler) Handle(ctx context.Context, cmd PlaceOrderCommand) (string, error) { // Create a new order order := models.Order{ ID: uuid.New().String(), UserID: cmd.UserID, Status: models.OrderStatusPending, Items: cmd.Items, CreatedAt: time.Now().UTC(), UpdatedAt: time.Now().UTC(), }

// Calculate total
order.CalculateTotal()

// Create an OrderPlaced event
event := events.NewOrderPlacedEvent(order, cmd.UserEmail)

// Store the event
if err := h.eventStore.Append(ctx, event); err != nil {
    return "", fmt.Errorf("failed to store order placed event: %w", err)
}

return order.ID, nil

}

Event Processors

// internal/processors/order_processor.go package processors

import ( "context" "encoding/json" "fmt" "log"

"github.com/yourusername/event-driven-app/internal/events"
"github.com/yourusername/event-driven-app/internal/models"
"github.com/yourusername/event-driven-app/internal/store"

)

type OrderProcessor struct { orderRepository store.OrderRepository eventStore store.EventStore }

func NewOrderProcessor(orderRepository store.OrderRepository, eventStore store.EventStore) *OrderProcessor { return &OrderProcessor{ orderRepository: orderRepository, eventStore: eventStore, } }

func (p *OrderProcessor) ProcessOrderPlaced(ctx context.Context, event events.Event) error { var payload events.OrderPlacedPayload

// Convert the payload
payloadJSON, err := json.Marshal(event.Payload)
if err != nil {
    return fmt.Errorf("failed to marshal payload: %w", err)
}

if err := json.Unmarshal(payloadJSON, &payload); err != nil {
    return fmt.Errorf("failed to unmarshal OrderPlacedPayload: %w", err)
}

// Store the order in the read model
if err := p.orderRepository.Save(ctx, payload.Order); err != nil {
    return fmt.Errorf("failed to save order: %w", err)
}

// Send notification email
log.Printf("Sending order confirmation email to %s for order %s", 
            payload.UserEmail, payload.Order.ID)

return nil

}

func (p *OrderProcessor) ProcessOrderPaid(ctx context.Context, event events.Event) error { var payload events.OrderPaidPayload

// Convert the payload
payloadJSON, err := json.Marshal(event.Payload)
if err != nil {
    return fmt.Errorf("failed to marshal payload: %w", err)
}

if err := json.Unmarshal(payloadJSON, &payload); err != nil {
    return fmt.Errorf("failed to unmarshal OrderPaidPayload: %w", err)
}

// Retrieve the order
order, err := p.orderRepository.FindByID(ctx, payload.OrderID)
if err != nil {
    return fmt.Errorf("failed to find order: %w", err)
}

// Update order status
order.Status = models.OrderStatusPaid
order.UpdatedAt = event.Timestamp

// Save updated order
if err := p.orderRepository.Save(ctx, *order); err != nil {
    return fmt.Errorf("failed to update order: %w", err)
}

log.Printf("Order %s marked as paid", payload.OrderID)

return nil

}

Advanced Patterns and Considerations

Event Replay and Rebuilding State

One of the powerful features of event sourcing is the ability to replay events to rebuild state:

func (s *OrderService) RebuildOrderState(ctx context.Context, orderID string) (*models.Order, error) { // Load all events for this order events, err := s.eventStore.Load(ctx, "order", orderID) if err != nil { return nil, err }

// Start with an empty order
var order models.Order

// Apply each event in sequence
for _, event := range events {
    switch event.Type {
    case events.OrderPlaced:
        var payload events.OrderPlacedPayload
        if err := unmarshalPayload(event.Payload, &payload); err != nil {
            return nil, err
        }
        order = payload.Order
        
    case events.OrderPaid:
        var payload events.OrderPaidPayload
        if err := unmarshalPayload(event.Payload, &payload); err != nil {
            return nil, err
        }
        order.Status = models.OrderStatusPaid
        order.UpdatedAt = event.Timestamp
        
    // Handle other event types...
    }
}

return &order, nil

}

Scaling Event Processing

For high-throughput systems, you can scale event processing by:

1. Partitioning: Distribute events across multiple partitions
2. Consumer groups: Run multiple instances of each consumer
3. Parallel processing: Process multiple events concurrently

Here's an example of parallel processing with worker pools:

func startWorkerPool(ctx context.Context, consumer *kafka.Consumer, numWorkers int, processor *processors.OrderProcessor) { // Create job channel jobs := make(chan kafka.Message, 100)

// Start workers
var wg sync.WaitGroup
for i := 0; i < numWorkers; i++ {
    wg.Add(1)
    go worker(ctx, &wg, jobs, processor)
}

// Feed jobs to workers
go func() {
    for {
        select {
        case <-ctx.Done():
            close(jobs)
            return
        default:
            msg, err := consumer.Read(ctx)
            if err != nil {
                if ctx.Err() != nil {
                    close(jobs)
                    return
                }
                log.Printf("Error reading message: %v", err)
                continue
            }
            
            jobs <- msg
        }
    }
}()

// Wait for all workers to finish
wg.Wait()

}

func worker(ctx context.Context, wg *sync.WaitGroup, jobs <-chan kafka.Message, processor *processors.OrderProcessor) { defer wg.Done()

for {
    select {
    case <-ctx.Done():
        return
    case msg, ok := <-jobs:
        if !ok {
            return
        }
        
        // Process the message
        event, err := events.Deserialize(msg.Value)
        if err != nil {
            log.Printf("Error deserializing event: %v", err)
            continue
        }
        
        // Handle the event based on its type
        switch event.Type {
        case events.OrderPlaced:
            err = processor.ProcessOrderPlaced(ctx, event)
        case events.OrderPaid:
            err = processor.ProcessOrderPaid(ctx, event)
        // Handle other event types...
        default:
            log.Printf("Unknown event type: %s", event.Type)
        }
        
        if err != nil {
            log.Printf("Error processing event %s: %v", event.ID, err)
        }
    }
}

}

Handling Failed Events

It's important to handle failures gracefully in event processing. One approach is the Dead Letter Queue (DLQ) pattern:

func (p *OrderProcessor) ProcessEvent(ctx context.Context, event events.Event) error { var err error

// Try to process the event
switch event.Type {
case events.OrderPlaced:
    err = p.ProcessOrderPlaced(ctx, event)
case events.OrderPaid:
    err = p.ProcessOrderPaid(ctx, event)
// Handle other event types...
default:
    err = fmt.Errorf("unknown event type: %s", event.Type)
}

// If processing failed, send to DLQ
if err != nil {
    log.Printf("Failed to process event %s: %v", event.ID, err)
    
    // Add error information to metadata
    if event.Metadata == nil {
        event.Metadata = make(map[string]string)
    }
    event.Metadata["processing_error"] = err.Error()
    event.Metadata["failed_at"] = time.Now().UTC().Format(time.RFC3339)
    
    // Send to DLQ
    dlqEvent := events.NewEvent(
        "event.processing.failed",
        event.EntityType,
        event.EntityID,
        event, // The original event becomes the payload
        event.Metadata,
    )
    
    return p.eventStore.Append(ctx, dlqEvent)
}

return nil

}

Real-World Considerations

Monitoring and Observability

Monitoring is crucial for event-driven systems. Key metrics to track include:

1. Producer metrics: Event publish rate, latency, errors
2. Consumer metrics: Event processing rate, latency, errors
3. Kafka metrics: Consumer lag, partition distribution, broker health

Implement structured logging and tracing across services:

func (p *OrderProcessor) ProcessOrderPlaced(ctx context.Context, event events.Event) error { logger := log.With(). Str("event_id", event.ID). Str("event_type", string(event.Type)). Str("order_id", event.EntityID). Logger()

logger.Info().Msg("Processing OrderPlaced event")

// Process event...

logger.Info().Msg("OrderPlaced event processed successfully")
return nil

}

Testing Event-Driven Systems

Testing event-driven systems requires specialized approaches:

Unit Testing Event Handlers

Test that individual event handlers work correctly:

func TestOrderProcessor_ProcessOrderPlaced(t *testing.T) { // Create mocks mockRepo := &mocks.MockOrderRepository{} mockEventStore := &mocks.MockEventStore{}

// Create processor with mocks
processor := processors.NewOrderProcessor(mockRepo, mockEventStore)

// Create test order
order := models.Order{
    ID:     "order-123",
    UserID: "user-456",
    Status: models.OrderStatusPending,
    Items: []models.OrderItem{
        {ProductID: "prod-1", Quantity: 2, Price: 10.00},
    },
}
order.CalculateTotal() // Total should be 20.00

// Create test event
payload := events.OrderPlacedPayload{
    Order:     order,
    UserEmail: "user@example.com",
}

event := events.Event{
    ID:         "event-789",
    Type:       events.OrderPlaced,
    EntityType: "order",
    EntityID:   order.ID,
    Timestamp:  time.Now().UTC(),
    Version:    "1.0",
    Payload:    payload,
}

// Setup expectations
mockRepo.On("Save", mock.Anything, order).Return(nil)

// Process the event
err := processor.ProcessOrderPlaced(context.Background(), event)

// Assert expectations
assert.NoError(t, err)
mockRepo.AssertExpectations(t)

}

Integration Testing with Test Containers

Use the testcontainers library to spin up Kafka in tests:

func TestOrderService_Integration(t *testing.T) { if testing.Short() { t.Skip("Skipping integration test in short mode") }

// Start Kafka test container
ctx := context.Background()
kafkaContainer, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
    ContainerRequest: testcontainers.ContainerRequest{
        Image:        "confluentinc/cp-kafka:latest",
        ExposedPorts: []string{"9092/tcp"},
        Env: map[string]string{
            "KAFKA_ADVERTISED_LISTENERS": "PLAINTEXT://localhost:9092",
            "KAFKA_LISTENER_SECURITY_PROTOCOL_MAP": "PLAINTEXT:PLAINTEXT",
            "KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR": "1",
        },
        WaitingFor: wait.ForLog("started"),
    },
    Started: true,
})
if err != nil {
    t.Fatalf("Failed to start Kafka container: %v", err)
}
defer kafkaContainer.Terminate(ctx)

// Get Kafka address
mappedPort, err := kafkaContainer.MappedPort(ctx, "9092/tcp")
if err != nil {
    t.Fatalf("Failed to get mapped port: %v", err)
}
host, err := kafkaContainer.Host(ctx)
if err != nil {
    t.Fatalf("Failed to get host: %v", err)
}

kafkaAddress := fmt.Sprintf("%s:%s", host, mappedPort.Port())

// Create Kafka producer and consumer
producer := kafka.NewProducer([]string{kafkaAddress})
defer producer.Close()

consumer := kafka.NewConsumer(
    []string{kafkaAddress},
    "events.order",
    "test-group",
)
defer consumer.Close()

// Create event store
eventStore := store.NewKafkaEventStore(producer, consumer, "events.")

// Test publishing and consuming events
// ...

}

End-to-End Testing

Test the complete flow across multiple services:

func TestOrderFlow_E2E(t *testing.T) { if testing.Short() { t.Skip("Skipping E2E test in short mode") }

// Start services (often done with Docker Compose)
services := startTestServices(t)
defer services.Cleanup()

// Create a test order
orderCmd := commands.PlaceOrderCommand{
    UserID: "user-123",
    Items: []models.OrderItem{
        {ProductID: "product-1", Quantity: 2, Price: 10.00},
    },
    UserEmail: "test@example.com",
}

// Send command to place order
orderID, err := sendPlaceOrderCommand(services.OrderServiceURL, orderCmd)
require.NoError(t, err)

// Wait for order to be created (poll read model)
order := waitForOrder(t, services.OrderQueryURL, orderID, 5*time.Second)
assert.Equal(t, models.OrderStatusPending, order.Status)

// Send payment command
paymentCmd := commands.ProcessPaymentCommand{
    OrderID: orderID,
    Amount:  order.Total,
    PaymentMethod: "credit_card",
    CardDetails: commands.CardDetails{
        Number: "4111111111111111",
        Expiry: "12/25",
        CVV:    "123",
    },
}

paymentID, err := sendProcessPaymentCommand(services.PaymentServiceURL, paymentCmd)
require.NoError(t, err)

// Wait for order to be updated to paid status
paidOrder := waitForOrderStatus(t, services.OrderQueryURL, orderID, models.OrderStatusPaid, 5*time.Second)
assert.Equal(t, models.OrderStatusPaid, paidOrder.Status)

// Verify events were published correctly
events := getEventsForOrder(services.EventStoreURL, orderID)
assert.Len(t, events, 2) // OrderPlaced and OrderPaid

}

Performance Considerations

When building high-throughput event-driven systems, consider these performance aspects:

1. Batch Processing

Process messages in batches to improve throughput:

func (c *BatchingConsumer) ProcessBatch(ctx context.Context, maxBatchSize int, timeout time.Duration) error { var msgs []kafka.Message

// Collect messages up to maxBatchSize or until timeout
timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()

for len(msgs) < maxBatchSize {
    msg, err := c.reader.ReadMessage(timeoutCtx)
    if err != nil {
        if errors.Is(err, context.DeadlineExceeded) {
            break // Timeout reached, process what we have
        }
        return err
    }
    
    msgs = append(msgs, msg)
}

// No messages collected
if len(msgs) == 0 {
    return nil
}

// Process messages in batch
return c.processor.ProcessBatch(ctx, msgs)

}

2. Parallel Processing

Use worker pools to process events in parallel:

func StartWorkerPool(ctx context.Context, consumer kafka.Consumer, numWorkers int, processor EventProcessor) { jobs := make(chan kafka.Message, numWorkers10) var wg sync.WaitGroup

// Start workers
for i := 0; i < numWorkers; i++ {
    wg.Add(1)
    go worker(ctx, &wg, jobs, processor)
}

// Read messages and send to workers
go func() {
    defer close(jobs)
    for {
        select {
        case <-ctx.Done():
            return
        default:
            msg, err := consumer.Read(ctx)
            if err != nil {
                if ctx.Err() != nil {
                    return
                }
                log.Printf("Error reading message: %v", err)
                continue
            }
            
            select {
            case jobs <- msg:
                // Message sent to worker
            case <-ctx.Done():
                return
            }
        }
    }
}()

// Wait for workers to finish
wg.Wait()

}

3. Optimizing Serialization

Choose efficient serialization formats. While JSON is convenient, Protocol Buffers or Avro provide much better performance:

// Example Protocol Buffers definition syntax = "proto3"; package events; option go_package = "github.com/yourusername/event-driven-app/internal/events";

message OrderPlacedEvent { string id = 1; string order_id = 2; string user_id = 3; string user_email = 4; repeated OrderItem items = 5; double total = 6; int64 created_at = 7; }

message OrderItem { string product_id = 1; int32 quantity = 2; double price = 3; }

// Using Protocol Buffers func SerializeEvent(event *events.OrderPlacedEvent) ([]byte, error) { return proto.Marshal(event) }

func DeserializeEvent(data []byte) (*events.OrderPlacedEvent, error) { event := &events.OrderPlacedEvent{} err := proto.Unmarshal(data, event) return event, err }

Real-World Architecture Example

Let's examine a complete architecture for an e-commerce system using event-driven architecture with Go and Kafka:

System Components:

1. Order Service: Handles order creation and management
2. Inventory Service: Manages product inventory
3. Payment Service: Processes payments
4. Notification Service: Sends emails and notifications
5. Analytics Service: Generates reports and insights

Event Flow:

1. Customer places an order -> OrderPlaced event
2. Inventory Service reserves items -> InventoryReserved event
3. Payment Service processes payment -> PaymentProcessed event
4. Order Service updates order status -> OrderConfirmed event
5. Notification Service sends confirmation -> NotificationSent event
6. Inventory Service updates stock levels -> InventoryUpdated event
7. Analytics Service records metrics -> OrderAnalyticsUpdated event

Architecture Diagram:

+----------------+         +----------------+         +----------------+
|                |         |                |         |                |
| Order Service  |<------->|    Kafka      |<------->|   Inventory    |
|                |         |                |         |    Service     |
+----------------+         +----------------+         +----------------+
        ^                         ^                         ^
        |                         |                         |
        v                         v                         v
+----------------+         +----------------+         +----------------+
|                |         |                |         |                |
|    Payment     |<------->| Notification  |<------->|   Analytics    |
|    Service     |         |    Service    |         |    Service     |
|                |         |                |         |                |
+----------------+         +----------------+         +----------------+

Implementation Considerations:

1. Service Independence: Each service has its own database and can function independently
2. Fault Tolerance: Services can continue functioning even if other services are down
3. Event Schema Evolution: Use versioning and schema registry to manage changes
4. Monitoring: Implement comprehensive monitoring at each layer
5. Deployment: Use containers and orchestration (e.g., Kubernetes)

Conclusion

Event-driven architecture with Go and Kafka provides a powerful framework for building scalable, resilient, and loosely coupled systems. By using events as the primary means of communication between services, you can achieve greater flexibility, improved fault tolerance, and better scalability.

Key takeaways from this article:

1. Event Design: Carefully design your events with clear schemas, versioning, and meaningful metadata
2. Reliable Processing: Implement idempotent consumers and exactly-once processing patterns
3. Schema Evolution: Plan for change by using versioning and maintaining compatibility
4. Scalability: Use partitioning, consumer groups, and parallel processing for high-throughput systems
5. Testing: Develop comprehensive testing strategies for event-driven systems
6. Monitoring: Implement robust observability across your event streams and services

While event-driven architecture offers many benefits, it also introduces complexity in testing, debugging, and maintaining consistency across services. By following the patterns and practices outlined in this article, you can harness the power of event-driven architecture while mitigating its challenges.

In future articles, I'll explore more advanced topics such as event sourcing with CQRS, implementing sagas for distributed transactions, and building real-time analytics pipelines with Kafka Streams and Go.

---

About the author: I'm a software engineer with experience in systems programming and distributed systems. Over the past years, I've been designing and implementing distributed systems in Go, with a recent focus on event-driven architectures and stream processing.