Event-Driven Architecture #

Ada satu momen yang sering menjadi titik balik dalam desain sistem: ketika kamu menyadari bahwa menambahkan fitur baru selalu membutuhkan mengubah kode yang sudah ada. Layanan notifikasi baru? Ubah Order Service. Analytics baru? Ubah Payment Service. Fitur loyalty points? Ubah semua service yang relevan. Setiap penambahan fitur menyebar seperti air yang mengalir ke celah-celah — mengubah kode yang seharusnya tidak perlu disentuh. Event-Driven Architecture (EDA) membalik pola ini: ketika sesuatu terjadi di sistem, sebuah event dipublikasikan. Siapapun yang peduli bisa subscribe tanpa sepengetahuan publisher. Order Service tidak perlu tahu ada Notification Service. Payment Service tidak perlu tahu ada Analytics Service. Setiap service hanya mengerjakan tugasnya, mempublikasikan apa yang terjadi, dan membiarkan ekosistem bereaksi sesuai kebutuhannya masing-masing.

Tiga Pola Komunikasi dalam EDA #

EDA bukan satu cara tunggal — ada tiga pola yang berbeda berdasarkan seberapa banyak data yang dibawa event:

flowchart TD
    subgraph EN["Event Notification"]
        P1["Order Service"] -->|"OrderPlaced\n{order_id: 'abc'}"| B1["Broker"]
        B1 --> C1["Notification Service\n\n→ fetch detail via API"]
    end

    subgraph ECST["Event Carried State Transfer"]
        P2["Order Service"] -->|"OrderPlaced\n{order_id, customer_id,\nitems, total, address}"| B2["Broker"]
        B2 --> C2["Notification Service\n\n→ langsung proses\ntanpa API call"]
    end

    subgraph ES["Event Sourcing"]
        P3["Order Service"] -->|"semua event\nsebagai sumber kebenaran"| B3["Event Store"]
        B3 --> C3["Read Model\n\n→ dibangun dari\nreplay event"]
    end

Pola	Event membawa	Kebutuhan API call	Cocok untuk
Event Notification	Hanya ID atau sinyal	Ya — consumer fetch detail	Event yang frequent, payload kecil
Event Carried State Transfer	Data lengkap	Tidak — self-contained	Consumer butuh data segera, tanpa API dependency
Event Sourcing	Semua perubahan state	Tidak — state dibangun dari event	Audit trail, time travel, CQRS

Komponen Utama EDA #

flowchart LR
    subgraph PROD["Producers"]
        OS["Order Service"]
        PS["Payment Service"]
        US["User Service"]
    end

    subgraph BROKER["Event Broker"]
        T1["topic: order.events"]
        T2["topic: payment.events"]
        T3["topic: user.events"]
    end

    subgraph CONS["Consumers"]
        NS["Notification Service"]
        AS["Analytics Service"]
        IS["Inventory Service"]
        LS["Loyalty Service"]
    end

    OS -->|"OrderPlaced, OrderCancelled"| T1
    PS -->|"PaymentProcessed, PaymentFailed"| T2
    US -->|"UserRegistered, UserUpdated"| T3

    T1 --> NS & AS & IS & LS
    T2 --> NS & AS & LS
    T3 --> NS & AS

Tiga keuntungan langsung dari topologi ini:

Extensibility — menambahkan LoyaltyService baru tidak perlu mengubah OrderService atau PaymentService; cukup subscribe ke topic yang relevan.

Resilience — jika NotificationService down, OrderService tidak terpengaruh; event tetap tersimpan di broker dan akan diproses saat NotificationService kembali online.

Scalability — setiap consumer bisa di-scale secara independen berdasarkan beban masing-masing.

Desain Event yang Baik #

Event adalah kontrak antara producer dan consumer. Desain yang buruk bisa mempersulit evolusi sistem.

// ✓ BENAR: Event dengan struktur yang jelas dan stabil
// Nama past tense, immutable, membawa data yang cukup

package events

import "time"

// OrderPlaced dipublikasikan ketika order berhasil dibuat
// Versi 1 — semua field backward compatible
type OrderPlaced struct {
	// Metadata event — wajib ada di semua event
	EventID    string    `json:"event_id"`    // ID unik event (untuk idempotency)
	EventName  string    `json:"event_name"`  // "order.placed"
	Version    int       `json:"version"`     // 1
	OccurredAt time.Time `json:"occurred_at"`

	// Payload event
	OrderID    string      `json:"order_id"`
	CustomerID string      `json:"customer_id"`
	Items      []OrderItem `json:"items"`
	TotalCents int64       `json:"total_cents"`
	Currency   string      `json:"currency"`
}

type OrderItem struct {
	ProductID  string `json:"product_id"`
	Name       string `json:"name"`
	Quantity   int    `json:"quantity"`
	PriceCents int64  `json:"price_cents"`
}

// PaymentProcessed dipublikasikan ketika payment berhasil
type PaymentProcessed struct {
	EventID       string    `json:"event_id"`
	EventName     string    `json:"event_name"` // "payment.processed"
	Version       int       `json:"version"`
	OccurredAt    time.Time `json:"occurred_at"`

	PaymentID  string `json:"payment_id"`
	OrderID    string `json:"order_id"`
	AmountCents int64 `json:"amount_cents"`
	Method     string `json:"method"` // "credit_card", "bank_transfer"
}

// ✗ ANTI-PATTERN: event dengan nama imperatif dan data minimal
type CreateOrder struct { // ✗ nama command, bukan event
	ID string // ✗ hanya ID — consumer harus API call untuk mendapat detail
}

Implementasi Producer: Publish ke Kafka #

// pkg/kafka/producer.go — generic Kafka producer
package kafka

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

	"github.com/segmentio/kafka-go"
)

// Producer adalah wrapper Kafka writer yang type-safe
type Producer struct {
	writer *kafka.Writer
}

func NewProducer(brokers []string, topic string) *Producer {
	return &Producer{
		writer: &kafka.Writer{
			Addr:         kafka.TCP(brokers...),
			Topic:        topic,
			Balancer:     &kafka.LeastBytes{},
			WriteTimeout: 5 * time.Second,
			RequiredAcks: kafka.RequireAll, // ✓ tunggu semua replica
		},
	}
}

// Publish mempublikasikan event ke Kafka
func (p *Producer) Publish(ctx context.Context, key string, event interface{}) error {
	payload, err := json.Marshal(event)
	if err != nil {
		return fmt.Errorf("gagal marshal event: %w", err)
	}

	err = p.writer.WriteMessages(ctx, kafka.Message{
		Key:   []byte(key),
		Value: payload,
		Headers: []kafka.Header{
			{Key: "content-type", Value: []byte("application/json")},
		},
	})
	if err != nil {
		return fmt.Errorf("gagal publish ke kafka: %w", err)
	}
	return nil
}

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

// internal/order/service.go — Order Service mempublikasikan event
package order

import (
	"context"
	"time"

	"myapp/events"
	"myapp/pkg/kafka"
)

type EventPublisher interface {
	Publish(ctx context.Context, key string, event interface{}) error
}

type OrderService struct {
	repo      OrderRepository
	publisher EventPublisher
}

func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) (*Order, error) {
	order := buildOrder(input)

	if err := s.repo.Save(ctx, order); err != nil {
		return nil, err
	}

	// Publikasikan event SETELAH berhasil disimpan
	// ✓ Event Carried State Transfer — membawa semua data yang dibutuhkan consumer
	event := events.OrderPlaced{
		EventID:    generateEventID(),
		EventName:  "order.placed",
		Version:    1,
		OccurredAt: time.Now(),
		OrderID:    order.ID,
		CustomerID: order.CustomerID,
		Items:      toEventItems(order.Items),
		TotalCents: order.TotalCents,
		Currency:   "IDR",
	}

	// Key = OrderID untuk memastikan event order yang sama masuk ke partisi yang sama
	// ✓ Menjaga ordering event per order
	if err := s.publisher.Publish(ctx, order.ID, event); err != nil {
		// Log error tapi jangan gagalkan order — gunakan outbox pattern untuk reliability
		slog.ErrorContext(ctx, "gagal publish OrderPlaced event",
			"order_id", order.ID,
			"error", err.Error(),
		)
	}

	return order, nil
}

Implementasi Consumer dengan Idempotency #

Consumer yang baik harus idempotent — memproses event yang sama lebih dari sekali tidak boleh menghasilkan efek yang berbeda. Kafka at-least-once delivery memastikan event bisa terkirim lebih dari sekali.

// internal/notification/consumer.go — Notification Service consumer
package notification

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

	"github.com/segmentio/kafka-go"

	"myapp/events"
)

// ProcessedEventRepository untuk menyimpan event yang sudah diproses
type ProcessedEventRepository interface {
	IsProcessed(ctx context.Context, eventID string) (bool, error)
	MarkProcessed(ctx context.Context, eventID string) error
}

type Consumer struct {
	reader       *kafka.Reader
	notifier     Notifier
	processedRepo ProcessedEventRepository
}

func NewConsumer(brokers []string, topic, groupID string, notifier Notifier, processedRepo ProcessedEventRepository) *Consumer {
	return &Consumer{
		reader: kafka.NewReader(kafka.ReaderConfig{
			Brokers:        brokers,
			Topic:          topic,
			GroupID:        groupID,        // ✓ consumer group untuk load balancing
			MinBytes:       10e3,           // 10KB
			MaxBytes:       10e6,           // 10MB
			CommitInterval: time.Second,    // auto commit setiap 1 detik
		}),
		notifier:     notifier,
		processedRepo: processedRepo,
	}
}

// Start memulai consumer loop
func (c *Consumer) Start(ctx context.Context) error {
	for {
		msg, err := c.reader.FetchMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil // context cancelled — shutdown normal
			}
			slog.Error("gagal fetch message", "error", err)
			continue
		}

		if err := c.processMessage(ctx, msg); err != nil {
			slog.Error("gagal proses message",
				"offset", msg.Offset,
				"error", err,
			)
			// Tidak commit offset — message akan diproses ulang
			continue
		}

		// Commit hanya setelah berhasil diproses
		if err := c.reader.CommitMessages(ctx, msg); err != nil {
			slog.Error("gagal commit offset", "error", err)
		}
	}
}

func (c *Consumer) processMessage(ctx context.Context, msg kafka.Message) error {
	var event events.OrderPlaced
	if err := json.Unmarshal(msg.Value, &event); err != nil {
		// Event tidak valid — skip (tidak ada gunanya retry)
		slog.Warn("event tidak valid, skip", "error", err)
		return nil
	}

	// ✓ IDEMPOTENCY CHECK — cek apakah event sudah pernah diproses
	processed, err := c.processedRepo.IsProcessed(ctx, event.EventID)
	if err != nil {
		return err
	}
	if processed {
		slog.Info("event sudah diproses, skip", "event_id", event.EventID)
		return nil // idempotent — tidak ada efek samping kedua kali
	}

	// Proses event
	if err := c.notifier.SendOrderConfirmation(ctx, event); err != nil {
		return err
	}

	// Tandai sudah diproses — SETELAH berhasil
	return c.processedRepo.MarkProcessed(ctx, event.EventID)
}

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

Dead Letter Queue: Menangani Event yang Gagal #

Tidak semua event bisa diproses berhasil — data korup, dependency tidak tersedia, atau bug di consumer. Dead Letter Queue (DLQ) adalah safety net untuk event yang gagal setelah N retry:

// pkg/kafka/dlq_consumer.go — Consumer dengan DLQ support
package kafka

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

	kgo "github.com/segmentio/kafka-go"
)

type ProcessFunc func(ctx context.Context, msg kgo.Message) error

type DLQConsumer struct {
	reader      *kgo.Reader
	dlqWriter   *kgo.Writer // writer ke DLQ topic
	process     ProcessFunc
	maxRetries  int
}

func NewDLQConsumer(
	brokers []string,
	topic, groupID, dlqTopic string,
	process ProcessFunc,
	maxRetries int,
) *DLQConsumer {
	return &DLQConsumer{
		reader: kgo.NewReader(kgo.ReaderConfig{
			Brokers: brokers,
			Topic:   topic,
			GroupID: groupID,
		}),
		dlqWriter: &kgo.Writer{
			Addr:  kgo.TCP(brokers...),
			Topic: dlqTopic,
		},
		process:    process,
		maxRetries: maxRetries,
	}
}

func (c *DLQConsumer) Start(ctx context.Context) error {
	for {
		msg, err := c.reader.FetchMessage(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			continue
		}

		var lastErr error
		for attempt := 1; attempt <= c.maxRetries; attempt++ {
			lastErr = c.process(ctx, msg)
			if lastErr == nil {
				break
			}

			slog.Warn("gagal proses event, retry",
				"attempt", attempt,
				"max_retries", c.maxRetries,
				"error", lastErr,
			)

			// Exponential backoff sebelum retry
			if attempt < c.maxRetries {
				select {
				case <-ctx.Done():
					return nil
				case <-time.After(time.Duration(attempt*attempt) * 100 * time.Millisecond):
				}
			}
		}

		if lastErr != nil {
			// ✓ Kirim ke DLQ setelah semua retry gagal
			dlqMsg := c.buildDLQMessage(msg, lastErr)
			if err := c.dlqWriter.WriteMessages(ctx, dlqMsg); err != nil {
				slog.Error("gagal kirim ke DLQ", "error", err)
				continue // jangan commit — coba lagi nanti
			}
			slog.Error("event dikirim ke DLQ",
				"topic", msg.Topic,
				"offset", msg.Offset,
				"error", lastErr,
			)
		}

		c.reader.CommitMessages(ctx, msg)
	}
}

func (c *DLQConsumer) buildDLQMessage(original kgo.Message, err error) kgo.Message {
	type DLQEnvelope struct {
		OriginalTopic   string    `json:"original_topic"`
		OriginalOffset  int64     `json:"original_offset"`
		OriginalPayload string    `json:"original_payload"`
		ErrorMessage    string    `json:"error_message"`
		FailedAt        time.Time `json:"failed_at"`
	}

	envelope, _ := json.Marshal(DLQEnvelope{
		OriginalTopic:   original.Topic,
		OriginalOffset:  original.Offset,
		OriginalPayload: string(original.Value),
		ErrorMessage:    err.Error(),
		FailedAt:        time.Now(),
	})

	return kgo.Message{
		Key:   original.Key,
		Value: envelope,
		Headers: append(original.Headers, kgo.Header{
			Key:   "x-dlq-reason",
			Value: []byte(err.Error()),
		}),
	}
}

Outbox Pattern: Garanteed Event Delivery #

Masalah umum di EDA: bagaimana memastikan event terkirim jika service crash setelah commit database tapi sebelum publish ke broker? Outbox Pattern menyelesaikan ini:

// Outbox pattern: simpan event ke database dalam satu transaksi dengan operasi utama
// Goroutine terpisah (relay) membaca outbox dan publish ke broker

// internal/order/service.go — dengan outbox pattern
func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) (*Order, error) {
	order := buildOrder(input)
	event := buildOrderPlacedEvent(order)

	// ✓ Satu transaksi database: simpan order DAN event ke outbox
	err := s.db.WithTransaction(ctx, func(tx *sql.Tx) error {
		// Simpan order
		if err := s.repo.SaveTx(ctx, tx, order); err != nil {
			return err
		}

		// Simpan event ke outbox table — dalam transaksi yang sama
		eventPayload, _ := json.Marshal(event)
		_, err := tx.ExecContext(ctx,
			`INSERT INTO outbox_events (id, topic, key, payload, created_at, published)
			 VALUES ($1, $2, $3, $4, NOW(), FALSE)`,
			event.EventID, "order.events", order.ID, eventPayload,
		)
		return err
	})
	if err != nil {
		return nil, err
	}

	return order, nil
}

// internal/outbox/relay.go — goroutine yang publish event dari outbox ke Kafka
type OutboxRelay struct {
	db        *sql.DB
	publisher EventPublisher
	interval  time.Duration
}

func (r *OutboxRelay) Start(ctx context.Context) {
	ticker := time.NewTicker(r.interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			r.publishPendingEvents(ctx)
		}
	}
}

func (r *OutboxRelay) publishPendingEvents(ctx context.Context) {
	// Ambil event yang belum dipublikasikan
	rows, err := r.db.QueryContext(ctx,
		`SELECT id, topic, key, payload FROM outbox_events
		 WHERE published = FALSE
		 ORDER BY created_at
		 LIMIT 100`,
	)
	if err != nil {
		return
	}
	defer rows.Close()

	for rows.Next() {
		var id, topic, key string
		var payload []byte
		rows.Scan(&id, &topic, &key, &payload)

		// Publish ke Kafka
		if err := r.publisher.PublishRaw(ctx, topic, key, payload); err != nil {
			slog.Error("gagal publish outbox event", "id", id, "error", err)
			continue
		}

		// Tandai sebagai published
		r.db.ExecContext(ctx,
			`UPDATE outbox_events SET published = TRUE, published_at = NOW() WHERE id = $1`,
			id,
		)
	}
}

sequenceDiagram
    participant OS as Order Service
    participant DB as Database
    participant OR as Outbox Relay
    participant KB as Kafka Broker
    participant NS as Notification Service

    OS->>DB: BEGIN TRANSACTION
    OS->>DB: INSERT orders (order data)
    OS->>DB: INSERT outbox_events (event payload, published=FALSE)
    OS->>DB: COMMIT
    Note over OS,DB: Atomik — keduanya berhasil atau keduanya gagal

    OR->>DB: SELECT outbox WHERE published=FALSE
    DB-->>OR: [event1, event2]
    OR->>KB: Publish event1
    KB-->>OR: ACK
    OR->>DB: UPDATE outbox SET published=TRUE

    KB->>NS: Consume event1
    NS->>NS: Proses notifikasi

Choreography vs Orchestration #

Ada dua pendekatan untuk mengkoordinasikan alur lintas service di EDA:

flowchart TD
    subgraph CHOREO["Choreography — Setiap service bereaksi pada event"]
        OS1["Order Service"] -->|"OrderPlaced"| MB1["Broker"]
        MB1 --> PS1["Payment Service"]
        PS1 -->|"PaymentProcessed"| MB1
        MB1 --> IS1["Inventory Service"]
        IS1 -->|"InventoryReserved"| MB1
        MB1 --> NS1["Notification Service"]
        Note1["✓ Loose coupling\n✓ Mudah tambah service baru\n✗ Alur bisnis tersebar\n✗ Sulit debug end-to-end"]
    end

    subgraph ORCH["Orchestration — Satu orchestrator mengontrol alur"]
        SAGA["Order Saga\nOrchestrator"]
        SAGA -->|"ProcessPayment"| PS2["Payment Service"]
        PS2 -->|"PaymentResult"| SAGA
        SAGA -->|"ReserveInventory"| IS2["Inventory Service"]
        IS2 -->|"InventoryResult"| SAGA
        SAGA -->|"SendNotification"| NS2["Notification Service"]
        Note2["✓ Alur bisnis jelas\n✓ Mudah handle failure\n✗ Orchestrator bisa jadi bottleneck\n✗ Coupling ke orchestrator"]
    end

Aspek	Choreography	Orchestration
Coupling	Sangat rendah	Sedang (ke orchestrator)
Visibilitas alur	Tersebar	Terpusat dan jelas
Error handling	Tiap service tanggung jawab sendiri	Orchestrator yang menangani
Kompleksitas	Tinggi saat flow rumit	Lebih terkelola untuk flow panjang
Cocok untuk	Alur sederhana, banyak consumer independen	Bisnis workflow yang panjang dan stateful

Event Versioning dan Schema Evolution #

Event adalah kontrak jangka panjang. Schema harus berevolusi dengan backward compatibility:

// events/order_placed.go — versioning event schema
package events

// V1 — initial schema
type OrderPlacedV1 struct {
	EventID    string    `json:"event_id"`
	Version    int       `json:"version"`    // = 1
	OrderID    string    `json:"order_id"`
	CustomerID string    `json:"customer_id"`
	TotalCents int64     `json:"total_cents"`
	OccurredAt time.Time `json:"occurred_at"`
}

// V2 — tambah field baru (backward compatible: field baru opsional)
type OrderPlacedV2 struct {
	EventID      string    `json:"event_id"`
	Version      int       `json:"version"`    // = 2
	OrderID      string    `json:"order_id"`
	CustomerID   string    `json:"customer_id"`
	TotalCents   int64     `json:"total_cents"`
	OccurredAt   time.Time `json:"occurred_at"`
	// Field baru — consumer lama akan ignore ini
	DiscountCode string    `json:"discount_code,omitempty"`
	DeliveryAddr string    `json:"delivery_address,omitempty"`
}

// Consumer yang mendukung multiple versi
func processOrderPlaced(ctx context.Context, payload []byte) error {
	// Decode versi dulu
	var meta struct {
		Version int `json:"version"`
	}
	json.Unmarshal(payload, &meta)

	switch meta.Version {
	case 1:
		var event OrderPlacedV1
		json.Unmarshal(payload, &event)
		return handleV1(ctx, event)
	case 2:
		var event OrderPlacedV2
		json.Unmarshal(payload, &event)
		return handleV2(ctx, event)
	default:
		slog.Warn("unknown event version", "version", meta.Version)
		return nil // skip — jangan panic untuk versi yang tidak dikenal
	}
}

Anti-Pattern yang Harus Dihindari #

// ✗ Event dikirim SEBELUM database commit — bisa inkonsisten
func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) error {
	order := buildOrder(input)

	// ✗ Publish dulu baru simpan — jika Save gagal, event sudah terkirim
	s.publisher.Publish(ctx, order.ID, OrderPlaced{OrderID: order.ID})
	return s.repo.Save(ctx, order) // ✗ mungkin gagal setelah event terkirim
}

// ✓ Publish SETELAH database commit berhasil (atau gunakan outbox pattern)
func (s *OrderService) PlaceOrder(ctx context.Context, input PlaceOrderInput) error {
	order := buildOrder(input)
	if err := s.repo.Save(ctx, order); err != nil { // ✓ simpan dulu
		return err
	}
	s.publisher.Publish(ctx, order.ID, OrderPlaced{OrderID: order.ID}) // ✓ publish setelah sukses
	return nil
}

// ✗ Consumer tidak idempotent — duplikat event menyebabkan double processing
func (c *Consumer) handleOrderPlaced(ctx context.Context, event OrderPlaced) error {
	// ✗ langsung kirim notifikasi tanpa cek duplikat
	return c.mailer.Send(ctx, event.CustomerID, "Order Confirmed", buildBody(event))
}

// ✓ Consumer idempotent — cek event ID sebelum proses
func (c *Consumer) handleOrderPlaced(ctx context.Context, event OrderPlaced) error {
	if ok, _ := c.repo.IsProcessed(ctx, event.EventID); ok {
		return nil // ✓ sudah diproses, skip
	}
	if err := c.mailer.Send(ctx, event.CustomerID, "Order Confirmed", buildBody(event)); err != nil {
		return err
	}
	return c.repo.MarkProcessed(ctx, event.EventID) // ✓ tandai setelah sukses
}

// ✗ Business logic di broker atau event schema
// Event sebagai command (imperatif) yang mendikte apa yang harus dilakukan consumer
type ProcessPaymentCommand struct { // ✗ ini command, bukan event
	OrderID string
	Action  string // "process_payment" — logic di event
}

// ✓ Event hanya menyatakan apa yang terjadi — consumer yang memutuskan reaksinya
type OrderPlaced struct { // ✓ past tense, facts, no instructions
	OrderID    string
	CustomerID string
	TotalCents int64
}

Checklist Review Event-Driven Architecture #

DESAIN EVENT:
  □ Nama event menggunakan past tense (OrderPlaced, PaymentFailed)
  □ Setiap event punya EventID unik untuk idempotency
  □ Event memiliki field Version untuk schema evolution
  □ Event immutable — tidak dimodifikasi setelah dipublikasikan

PRODUCER:
  □ Event dipublikasikan SETELAH operasi database berhasil
  □ Outbox pattern digunakan untuk guaranteed delivery
  □ Partition key dipilih untuk menjaga ordering yang diperlukan

CONSUMER:
  □ Semua consumer idempotent — memproses event yang sama dua kali aman
  □ Deduplication key disimpan dan di-check sebelum proses
  □ Commit offset hanya setelah event berhasil diproses
  □ DLQ dikonfigurasi untuk event yang gagal setelah N retry

RELIABILITY:
  □ Retry dengan exponential backoff dikonfigurasi
  □ Dead Letter Queue dimonitor dan ada proses untuk replay
  □ Alert dikonfigurasi saat DLQ melebihi threshold

OBSERVABILITY:
  □ Correlation ID / trace ID dipropagasi di semua event
  □ Consumer lag dimonitor per topic dan consumer group
  □ Event processing time diukur dan di-alert jika lambat
  □ DLQ size dimonitor

SCHEMA EVOLUTION:
  □ Perubahan event selalu backward compatible (tambah field opsional, jangan hapus)
  □ Breaking change memerlukan versi baru (v2, v3)
  □ Consumer lama bisa handle event versi baru (ignore unknown field)

Ringkasan #

EDA membalik arah dependency — producer tidak tahu siapa yang akan mengonsumsi event-nya; menambahkan consumer baru tidak perlu mengubah producer sama sekali.

Ada tiga pola — Event Notification (hanya sinyal), Event Carried State Transfer (data lengkap), dan Event Sourcing (event sebagai sumber kebenaran); pilih berdasarkan kebutuhan consumer.

Nama event past tense, payload immutable — event merepresentasikan fakta yang telah terjadi, bukan instruksi apa yang harus dilakukan; ini perbedaan fundamental antara event dan command.

Publish SETELAH commit database, bukan sebelum — event yang terkirim sebelum data tersimpan menyebabkan inkonsistensi; gunakan outbox pattern untuk guaranteed delivery.

Semua consumer harus idempotent — Kafka at-least-once delivery memastikan event bisa terkirim lebih dari sekali; consumer yang tidak idempotent akan menyebabkan duplikasi data.

DLQ adalah safety net yang wajib ada — event yang terus gagal harus diarahkan ke DLQ agar tidak memblokir event lain; DLQ harus dimonitor dan ada proses replay.

Outbox pattern untuk reliability — simpan event di database dalam satu transaksi dengan operasi utama; relay terpisah yang mempublikasikan ke broker.

Idempotency key harus disimpan — gunakan EventID sebagai key; sebelum proses, cek apakah event sudah pernah diproses; tandai setelah berhasil.

Event versioning untuk evolution — tambah field baru dengan omitempty; jangan hapus field; consumer lama harus bisa ignore field yang tidak dikenal.

Observability wajib lebih dari sistem sync — distributed tracing lintas event, consumer lag monitoring, dan DLQ alerting harus ada sebelum EDA masuk production.

← Sebelumnya: Service-Based Architecture Berikutnya: CQRS Architecture →