Replay Capability #

Dalam arsitektur sistem berbasis peristiwa (event-driven architecture), salah satu kebutuhan yang sering kali muncul adalah kemampuan untuk memproses ulang pesan-pesan yang telah dikirimkan di masa lalu. Skenario ini umumnya dipicu oleh berbagai kebutuhan operasional, seperti terjadinya kegagalan massal (disaster recovery) pada database aplikasi konsumen yang mengharuskan kita memulihkan data dari waktu tertentu, kebutuhan untuk melatih ulang model kecerdasan buatan (machine learning retraining) menggunakan dataset historis, atau ketika kita ingin menambahkan microservice baru yang harus membangun kondisi state internalnya sendiri berdasarkan riwayat transaksi sejak awal waktu.

Kemampuan untuk memutar kembali waktu pembacaan pesan ini secara industri dikenal sebagai Event Replay atau Time-Travel Reading. Di sinilah letak perbedaan yang sangat kontras antara filosofi antrean dinamis RabbitMQ dan log kekal Apache Kafka. Sementara satu platform dirancang untuk melupakan masa lalu demi efisiensi masa kini, platform lainnya justru mendokumentasikan setiap peristiwa secara kekal sebagai sumber kebenaran utama. Artikel ini akan membedah secara mendalam perbandingan kemampuan replay antara Kafka dan RabbitMQ.

Mekanisme Replay di Apache Kafka (Native Log Retention) #

Apache Kafka menyediakan kemampuan Event Replay secara native di tingkat arsitektur intinya. Karena Kafka memperlakukan setiap partisi topik sebagai Append-Only Immutable Log yang disimpan di dalam disk sekuensial, data tidak akan dihapus setelah dibaca oleh konsumen.

Setiap pesan di dalam partisi diidentifikasi dengan nomor indeks linear yang unik dan berurutan yang disebut Offset. Ketika aplikasi konsumen membaca data, konsumen hanya perlu memindahkan penunjuk offset membacanya.

[ Awal Log ] ──────────────────────────────────────────────> [ Akhir Log ]
 Offset 0     Offset 1     Offset 2     Offset 3     Offset 4
┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐
│ Pesan A  │ │ Pesan B  │ │ Pesan C  │ │ Pesan D  │ │ Pesan E  │
└──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘
                                           ▲
                                           │ Offset Saat Ini: 3 (Pesan D)
                                           │
                        [ REWIND OFFSET ] ─┘ Pindah Mundur ke Offset 1 (Pesan B)

Untuk memutar ulang pemrosesan data ke masa lalu, kita tidak perlu memodifikasi pesan di broker atau meminta produsen mengirim ulang data. Konsumen hanya perlu mengirimkan instruksi ke broker Kafka untuk memindahkan posisi offset membacanya mundur (offset rewind).

Kafka mendukung beberapa metode penentuan titik awal pembacaan ulang offset:

• earliest: Memaksa konsumen untuk memutar kembali pembacaan ke offset 0 (pesan paling awal yang masih tersimpan di disk retensi).
• latest: Mengabaikan seluruh data historis dan hanya mendengarkan pesan baru yang masuk setelah konsumen aktif.
• Manual Offset Seek: Memindahkan offset konsumen ke angka indeks spesifik secara presisi (misalnya melompat langsung ke offset 15000 untuk memproses ulang data dari titik tersebut).
• Timestamp-Based Seek: Meminta broker mencari offset terkecil yang ditulis pada waktu tertentu (misalnya: “cari offset yang direkam tepat pada tanggal 9 Juni 2026 pukul 09.00”). Ini sangat praktis untuk memulihkan kegagalan database yang terjadi sejak waktu spesifik.

Kebijakan Retensi Log Kafka (Log Retention Policy) #

Meskipun log Kafka bersifat kekal, kita tidak bisa menyimpan seluruh data transaksi dunia selamanya di satu server disk tanpa batas. Kafka mengelola kapasitas penyimpanan disk menggunakan kebijakan retensi (Retention Policy) yang dapat dikonfigurasi secara fleksibel per topik:

1. Retensi Berbasis Waktu (Time-based Retention) #

Menggunakan parameter log.retention.hours atau log.retention.ms. Kebijakan ini menentukan berapa lama pesan boleh mengendap di disk sebelum dihapus otomatis. Contoh: jika disetel 7 hari (default), data transaksi berumur 8 hari akan dihapus secara bertahap oleh broker dari segmen log terlama, tidak peduli apakah data tersebut sudah dibaca oleh seluruh konsumen atau belum.

2. Retensi Berbasis Ukuran (Size-based Retention) #

Menggunakan parameter log.retention.bytes. Kebijakan ini membatasi ukuran file log partisi maksimal pada kapasitas tertentu (misal 50 GB). Jika ukuran file melampaui batas tersebut karena derasnya aliran data produsen, Kafka akan menghapus segmen log tertua untuk menjaga kapasitas disk tetap aman.

3. Pemadatan Log (Log Compaction) #

Menggunakan kebijakan log.cleanup.policy = compact. Ini adalah fitur canggih di mana Kafka akan memindai berkas log dan hanya mempertahankan pesan dengan nilai payload terbaru (latest value) untuk setiap kunci partisi (partition key) unik, menghapus versi history yang lebih lama.

Pola ini sangat ideal untuk mereplikasi status database (Change Data Capture / CDC), di mana kita hanya peduli pada status saldo atau profil pengguna teraktual, bukan histori mutasi perubahannya setiap detik.

Batasan Replay di RabbitMQ #

Model antrean tradisional RabbitMQ yang menganut prinsip destructive read menghapus pesan dari RAM dan disk begitu konsumen mengirimkan ACK. Hal ini membuat RabbitMQ biasa tidak mendukung kemampuan Event Replay secara native.

Jika kita membutuhkan pemrosesan ulang data gagal di RabbitMQ, kita harus memilih salah satu dari solusi alternatif berikut:

1. Republishing dari Database Eksternal (Taktik Manual) #

Karena RabbitMQ tidak menyimpan histori pesan, tim developer harus merancang aplikasi untuk menulis setiap payload pesan yang diterbitkan produsen ke dalam database relasional (seperti PostgreSQL) atau NoSQL (seperti MongoDB) terlebih dahulu sebagai log audit.

Jika terjadi bencana database pada konsumen dan kita perlu melakukan replay data dari 3 jam yang lalu, kita harus menulis skrip migrasi khusus untuk membaca baris log dari database eksternal tersebut dan mempublikasikannya kembali (re-publishing) ke Exchange RabbitMQ secara manual. Taktik ini memakan waktu, rentan kesalahan, dan menambah beban I/O pada database log kita.

2. Pemanfaatan RabbitMQ Streams (Taktik Native Modern) #

Untuk menutupi kelemahan fatal ini dan bersaing dengan keunggulan Kafka, RabbitMQ sejak versi 3.9 memperkenalkan tipe antrean baru bernama RabbitMQ Streams.

Streams adalah implementasi log append-only kekal yang berjalan di dalam ekosistem RabbitMQ. Berbeda dengan antrean Classic atau Quorum, pesan di dalam antrean tipe Streams tidak dihapus setelah dikonsumsi. Status ACK konsumen diabaikan oleh broker untuk penghapusan fisik. Pesan disimpan secara sekuensial di disk dan dibatasi hanya menggunakan retensi berbasis waktu atau ukuran.

flowchart LR
    Consumer["Consumer"] -->|"Baca dari Offset X / Timestamp Y"| Streams["RabbitMQ Streams (Append-Only Log)"]

Konsumen yang terhubung ke RabbitMQ Streams dapat menggunakan protokol khusus (menggunakan port 5552 alih-alih port AMQP standar 5672) untuk melakukan penentuan posisi offset pembacaan awal, memutar mundur offset, atau membaca dari timestamp tertentu secara native mirip seperti Kafka. Hal ini memberikan solusi bagi tim yang membutuhkan fitur replay tetapi ingin tetap mempertahankan satu teknologi broker tunggal (RabbitMQ) di infrastruktur mereka.

Tabel Perbandingan Kemampuan Replay #

Berikut adalah tabel matriks komparatif yang menyandingkan kemampuan replay pada Kafka, RabbitMQ biasa (Classic/Quorum), dan RabbitMQ Streams:

Implementasi Kode Go (Golang) #

Untuk menunjukkan perbedaan teknis penanganan replay, berikut adalah contoh implementasi konsumen dalam bahasa Go untuk Apache Kafka (menggunakan kafka-go dengan pengaturan rewind offset ke awal log) dan konsumen RabbitMQ Streams (menggunakan driver resmi rabbitmq/stream-go-client).

1. Konsumer Kafka (Replay Data dari Awal / Earliest Offset) #

Di Kafka, kita menyetel StartOffset ke kafka.FirstOffset untuk memutar ulang seluruh data historis dari awal partisi log yang tersimpan.

package main

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

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

func main() {
	// Konfigurasi pembaca Kafka dengan setelan replay dari awal
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     []string{"localhost:9092"},
		Topic:       "audit-events",
		GroupID:     "audit-replay-group-v1", // Nama group baru untuk memicu pembacaan ulang
		StartOffset: kafka.FirstOffset,       // MEMUTAR ULANG DARI EVENT TERAWAL (Offset 0)
	})
	defer r.Close()

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	log.Println("[KAFKA] Mulai memproses ulang data audit dari awal log...")

	go func() {
		for {
			msg, err := r.ReadMessage(ctx)
			if err != nil {
				log.Printf("Gagal membaca event: %v", err)
				break
			}
			// Memproses ulang data historis
			log.Printf("[REPLAY] Offset: %d | Payload: %s", msg.Offset, string(msg.Value))
		}
	}()

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	<-sigChan
}

2. Konsumer RabbitMQ Streams (Replay Data dari Awal via Stream Protocol) #

Di RabbitMQ Streams, kita dapat menyetel offset awal pembacaan menggunakan tipe OffsetSpecification ke nilai First untuk memicu pembacaan dari event terawal secara native.

package main

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

	"github.com/rabbitmq/stream-go-client/pkg/amqp"
	"github.com/rabbitmq/stream-go-client/pkg/stream"
)

func main() {
	// 1. Membuka koneksi ke port Stream RabbitMQ (port default 5552)
	env, err := stream.NewEnvironment(
		stream.NewEnvironmentOptions().
			SetHost("localhost").
			SetPort(5552).
			SetUser("guest").
			SetPassword("guest"),
	)
	if err != nil {
		log.Fatalf("Gagal koneksi ke RabbitMQ Stream port: %v", err)
	}
	defer env.Close()

	// 2. Mendaftarkan konsumen pada antrean tipe stream
	// Kita menentukan OffsetSpecification ke First() untuk memicu replay data
	handle, err := env.NewConsumer(
		"audit.stream.queue",
		func(ctx stream.ConsumerContext, message *amqp.Message) {
			// Callback pemrosesan ulang pesan historis
			log.Printf("[STREAM REPLAY] Offset: %d | Payload: %s", 
				ctx.Consumer.GetLastConsumedOffset(), string(message.GetData()))
		},
		stream.NewConsumerOptions().
			SetOffset(stream.OffsetSpecification{}.First()), // REPLAY DARI AWAL
	)
	if err != nil {
		log.Fatalf("Gagal mendaftarkan stream consumer: %v", err)
	}
	defer handle.Close()

	log.Println("[RABBITMQ STREAM] Mulai membaca ulang stream audit dari awal...")

	sigChan := make(chan os.Signal, 1)
	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
	<-sigChan
}

Anti-Pattern vs Solusi Praktis #

Mencoba mensimulasikan kemampuan replay pada tipe antrean yang salah dapat mengakibatkan kelumpuhan klaster. Berikut adalah anti-pattern yang paling sering dijumpai:

Anti-Pattern: Menumpuk Pesan di Antrean Classic/Quorum RabbitMQ untuk Kebutuhan Replay Masa Depan #

Mendesain sistem di mana produsen mengirimkan pesan event historis ke antrean Classic atau Quorum RabbitMQ yang sengaja tidak memiliki konsumen aktif, dengan maksud menimbun data tersebut di memori broker agar suatu saat nanti (misal beberapa minggu ke depan) dapat dikonsumsi oleh service baru yang membutuhkan data historis.

Mengapa ini salah? #

Ini adalah penyalahgunaan fatal terhadap fungsi antrean RabbitMQ. Antrean Classic dan Quorum dirancang optimal untuk menangani antrean jangka pendek yang cepat dikosongkan.

Menimbun jutaan pesan di antrean aktif tanpa dikonsumsi akan:

1. Menghabiskan RAM Broker: Broker Erlang harus mempertahankan metadata indeks, status pencarian, dan payload pesan di RAM. Ketika batas ambang watermark memori terlampaui, broker akan membeku (blocked state) dan menghentikan seluruh aktivitas publishing produsen.
2. Degradasi Performa I/O: Ketika memori RAM penuh, broker RabbitMQ terpaksa melakukan proses paging (menulis pesan transient ke disk dan menghapusnya dari RAM) secara terus-menerus. Proses pembacaan data nantinya akan berjalan sangat lambat karena broker harus memanggil disk I/O secara acak (random read) untuk membaca data dari berkas segmen yang terfragmentasi.

Solusi Praktis #

Jika sistem kita memiliki kebutuhan yang mutlak untuk menyimpan data event mentah selama berbulan-bulan dan memerlukan kemampuan memutar ulang offset secara bebas, gunakan Apache Kafka sebagai platform pipa data utama. Jika infrastruktur kita dibatasi hanya boleh menggunakan satu teknologi broker RabbitMQ, gunakan tipe antrean khusus RabbitMQ Streams yang dioptimalkan secara native untuk penyimpanan disk linear kekal berlatensi rendah, bukan antrean Classic atau Quorum biasa.

Ringkasan #

• Filosofi Replay — Event Replay adalah kemampuan untuk memutar kembali waktu pembacaan pesan historis. Kafka mendukung fitur ini secara native, sedangkan RabbitMQ biasa menghapus pesan pasca ACK.
• Mekanisme Offset Kafka — Kafka melacak posisi baca menggunakan indeks Offset linear. Konsumen dapat melakukan replay dengan meminta broker memindahkan offset membacanya mundur (offset rewind) ke awal (earliest) atau titik tertentu secara bebas.
• Kebijakan Retensi Kafka — Log Kafka dihapus secara bertahap berdasarkan masa umur data (waktu) atau kapasitas penyimpanan (bytes), bukan berdasarkan status ACK konsumen.
• Fitur Log Compaction — Kafka dapat memadatkan log untuk mempertahankan hanya payload status teraktual per kunci partisi unik, sangat berguna untuk replikasi state.
• Alternatif RabbitMQ Streams — Sejak versi 3.9, RabbitMQ menyediakan tipe antrean Streams yang mengadopsi model append-only log kekal dan stream protocol (port 5552) untuk mendukung replay asinkron.
• Bahaya Timbunan Pesan — Jangan pernah menumpuk jutaan pesan tanpa konsumen di antrean Classic/Quorum RabbitMQ untuk tujuan replay karena akan memicu memory watermark alarm yang membekukan broker. Gunakan Kafka atau RabbitMQ Streams.

← Sebelumnya: Ordering Behavior   Berikutnya: Throughput & Scalability →