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
- Loose Coupling: Services don't need direct knowledge of each other
- Scalability: Components can scale independently
- Resilience: Failure in one component doesn't bring down the entire system
- Flexibility: Easier to adapt and extend the system
- 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:
- Event ID: Unique identifier for the event
- Event Type: Describes what happened
- Entity Type: The type of entity affected
- Entity ID: The identifier of the affected entity
- Timestamp: When the event occurred
- Version: Schema version for evolution
- Payload: The actual event data
- 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:
- Always add fields as optional
- Never remove fields (mark them as deprecated instead)
- Never change field types
- 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:
- At-most-once: Messages might be lost but never processed twice
- At-least-once: Messages are never lost but might be processed multiple times
- 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:
-
Within a database transaction:
- Update the application state
- Insert the event into an "outbox" table
-
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:
- Partitioning: Distribute events across multiple partitions
- Consumer groups: Run multiple instances of each consumer
- 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:
- Producer metrics: Event publish rate, latency, errors
- Consumer metrics: Event processing rate, latency, errors
- 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:
- Order Service: Handles order creation and management
- Inventory Service: Manages product inventory
- Payment Service: Processes payments
- Notification Service: Sends emails and notifications
- Analytics Service: Generates reports and insights
Event Flow:
- Customer places an order ->
OrderPlaced event
- Inventory Service reserves items ->
InventoryReserved event
- Payment Service processes payment ->
PaymentProcessed event
- Order Service updates order status ->
OrderConfirmed event
- Notification Service sends confirmation ->
NotificationSent event
- Inventory Service updates stock levels ->
InventoryUpdated event
- Analytics Service records metrics ->
OrderAnalyticsUpdated event
Architecture Diagram:
+----------------+ +----------------+ +----------------+
| | | | | |
| Order Service |<------->| Kafka |<------->| Inventory |
| | | | | Service |
+----------------+ +----------------+ +----------------+
^ ^ ^
| | |
v v v
+----------------+ +----------------+ +----------------+
| | | | | |
| Payment |<------->| Notification |<------->| Analytics |
| Service | | Service | | Service |
| | | | | |
+----------------+ +----------------+ +----------------+
Implementation Considerations:
- Service Independence: Each service has its own database and can function independently
- Fault Tolerance: Services can continue functioning even if other services are down
- Event Schema Evolution: Use versioning and schema registry to manage changes
- Monitoring: Implement comprehensive monitoring at each layer
- 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:
- Event Design: Carefully design your events with clear schemas, versioning, and meaningful metadata
- Reliable Processing: Implement idempotent consumers and exactly-once processing patterns
- Schema Evolution: Plan for change by using versioning and maintaining compatibility
- Scalability: Use partitioning, consumer groups, and parallel processing for high-throughput systems
- Testing: Develop comprehensive testing strategies for event-driven systems
- 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.
