Event-Driven Architecture dengan Kafka dan Go Workers

Dalam 15 menit ke depan, kita akan merakit sebuah sistem pemrosesan asinkron berkecepatan tinggi menggunakan Apache Kafka dan worker Go yang handal langsung dari dapur produksi. Lupakan sejenak definisi teoretis yang kaku, karena di lapangan, tantangan terbesarnya adalah bagaimana menjaga kestabilan aliran data saat traffic melonjak tiba-tiba tanpa membuat service kita megap-megap.

1. Menu Kita Hari Ini & Alat yang Perlu Disiapkan

Bayangkan kita sedang menangani platform e-commerce skala menengah yang menerima ribuan event 'checkout' per detik. Monolith backend kita mulai keteteran karena database utama kewalahan mencatat log transaksi secara sinkron. Solusinya? Kita pisahkan beban tersebut menggunakan Apache Kafka sebagai buffer pesan yang tangguh, lalu kita racik sekumpulan worker Go yang siap melahap antrean tersebut secara paralel.

Sebelum mulai meracik, pastikan di mesin lokal atau staging server Anda sudah terpasang:

• Go (versi 1.21 atau lebih baru)
• Docker dan Docker Compose (untuk menjalankan Kafka dan Zookeeper dengan cepat)
• Library Go segmentio/kafka-go atau confluentinc/confluent-kafka-go (kita akan pakai segmentio/kafka-go karena murni Go dan minim dependensi C)

[!IMPORTANT] Pastikan port 9092 untuk Kafka dan 2181 untuk Zookeeper tidak sedang digunakan oleh service lain di mesin Anda.

2. Peta Jalan Singkat: Apa yang Sedang Kita Rakit

Perjalanan 15 menit kita dibagi menjadi tiga fase utama yang krusial dalam arsitektur event-driven:

1. Partisioning dan Consumer Group: Bagaimana cara Kafka membagi beban data secara merata ke beberapa worker Go agar tidak ada bottleneck.
2. Manajemen Offset dan Retry Otomatis: Cara memastikan tidak ada pesan yang hilang (at-least-once delivery) dan bagaimana menangani error tanpa membuat worker hang.
3. Graceful Shutdown Worker Go: Mengakhiri proses worker dengan anggun, menyelesaikan sisa job yang sedang berjalan sebelum aplikasi benar-benar mati.

Mari kita siapkan fondasinya sekarang juga agar tangan kita langsung kotor dengan kode.

3. Langkah 1: Mempersiapkan Fondasi & Environment

Kita mulai dengan menyiapkan environment docker-compose untuk menjalankan Kafka. Buat sebuah file bernama docker-compose.yml di direktori kerja Anda:

version: '3.8'

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0

Jalankan perintah berikut di terminal Anda untuk menghidupkan mesin:

docker-compose up -d

[!TIP] Jalankan docker logs kafka untuk memastikan broker sudah siap menerima koneksi. Jika Anda melihat log [Kafka Server 1], started, itu tanda bahwa dapur kita sudah panas dan siap digunakan.

4. Langkah 2: Meracik Konfigurasi Inti Tanpa Basa-basi

Sekarang, mari inisialisasi project Go kita dan buat worker yang memanfaatkan Consumer Group agar beban kerja terdistribusi dengan sempurna. Buat file main.go:

package main

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

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

func main() {
	topic := "order-events"
	groupID := "order-worker-group"

	// Konfigurasi Reader dengan Consumer Group
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:  []string{"localhost:9092"},
		GroupID:  groupID,
		Topic:    topic,
		MinBytes: 10e3, // 10KB
		MaxBytes: 10e6, // 10MB
	})
	defer r.Close()

	log.Println("Worker Go siap mendengarkan event dari Kafka...")

	// Channel untuk menangkap sinyal sistem (Graceful Shutdown)
	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)

	ctx, cancel := context.WithCancel(context.Background())
	go func() {
		<-sigChan
		cancel()
	}()

	for {
		select {
		case <-ctx.Done():
			log.Println("Mematikan worker secara elegan...")
			return
		default:
			m, err := r.FetchMessage(ctx)
			if err != nil {
				log.Printf("Gagal mengambil pesan: %v\n", err)
				continue
			}

			// Proses pesan (Simulasi bisnis logic)
			fmt.Printf("Pesan diterima dari Partisi %d at offset %d: %s\n", m.Partition, m.Offset, string(m.Value))

			// Komit offset secara manual setelah sukses diproses
			if err := r.CommitMessages(ctx, m); err != nil {
				log.Printf("Gagal commit offset: %v\n", err)
			}
		}
	}
}

Kode di atas menerapkan konsep at-least-once delivery. Pesan hanya akan di-commit offset-nya setelah benar-benar berhasil dieksekusi oleh worker.

5. Langkah 3: Sentuhan Akhir Keamanan & Optimasi

Di lingkungan production, koneksi ke Kafka tidak boleh dibiarkan polos tanpa penanganan error dan backoff retry. Jika terjadi kegagalan database saat memproses event, worker kita tidak boleh langsung melakukan spamming retry yang berujung pada cascading failure.

Kita tambahkan logika penanganan error sederhana di dalam loop pemrosesan:

func processMessageWithRetry(ctx context.Context, m kafka.Message) error {
	maxRetries := 3
	for i := 0; i < maxRetries; i++ {
		err := businessLogic(m.Value)
		if err == nil {
			return nil
		}
		log.Printf("Percobaan %d gagal, mencoba ulang...\n", i+1)
	}
	// Jika tetap gagal setelah max retries, kirim ke Dead Letter Queue (DLK)
	return fmt.Errorf("gagal memproses pesan setelah %d percobaan", maxRetries)
}

func businessLogic(data []byte) error {
	// Lakukan operasi database atau HTTP call di sini
	return nil
}

Dengan memisahkan businessLogic dan menerapkan batas retry, infrastruktur backend kita terlindungi dari lonjakan beban yang tidak perlu.

6. Tes Rasa: Memastikan Semuanya Berjalan Sempurna

Mari kita uji racikan kita. Jalankan worker Go Anda di terminal pertama:

go run main.go

Buka terminal kedua, lalu gunakan utility bawaan Kafka untuk mengirim pesan uji coba ke topik order-events:

docker exec -it kafka kafka-console-producer --bootstrap-server localhost:9092 --topic order-events

Ketikkan beberapa string acak seperti {"order_id": 101, "item": "Sepatu"} lalu tekan Enter. Perhatikan terminal pertama Anda. Anda akan melihat log instan yang menandakan worker Go berhasil menangkap, memproses, dan meng-commit offset pesan tersebut dengan mulus.

7. Pertolongan Pertama: Jika Terjadi Kendala Kecil

• Error connection refused pada port 9092: Pastikan container Docker Kafka sudah benar-benar berjalan dan variabel KAFKA_ADVERTISED_LISTENERS diset ke PLAINTEXT://localhost:9092.
• Pesan dibaca berulang-ulang (Infinite Loop): Ini biasanya terjadi karena CommitMessages gagal dipanggil atau gagal terkoneksi kembali ke Zookeeper/Kafka broker. Pastikan konteks jaringan stabil.

Dengan arsitektur ini, sistem Anda kini siap menangani jutaan event harian layaknya sistem enterprise berkaliber tinggi.