Ordering Guarantee #

Di dalam sistem terdistribusi, menjaga urutan kronologis kejadian (event ordering) adalah salah satu tantangan paling rumit sekaligus krusial. Bayangkan sebuah sistem perbankan di mana transaksi deposit sebesar Rp1.000.000 ditulis setelah transaksi penarikan sebesar Rp500.000, padahal penarikan terjadi kemudian. Jika urutan pesan ini terbalik, sistem mungkin akan menolak transaksi penarikan karena mendeteksi saldo tidak mencukupi, yang berujung pada kegagalan transaksi dan ketidakpuasan pengguna. Apache Kafka dirancang untuk mengatasi masalah ini dengan menyediakan jaminan pengurutan yang sangat kuat, namun jaminan tersebut memiliki batasan-batasan arsitektural yang harus kita pahami dengan baik agar tidak menimbulkan bug di lingkungan produksi.

Jaminan Pengurutan Tingkat Partisi vs Topik #

Salah satu kesalahpahaman paling umum di kalangan arsitek perangkat lunak yang baru menggunakan Kafka adalah menganggap bahwa Kafka menjamin urutan pesan di tingkat topik secara keseluruhan (topic-level ordering). Pada kenyataannya, Kafka hanya menjamin urutan pesan di tingkat partisi (partition-level ordering).

Mengapa Topic-Level Ordering Merupakan Anti-Pattern? #

Untuk memahami mengapa Kafka membatasi jaminan pengurutan pada tingkat partisi, kita harus melihat kembali tujuan utama Kafka diciptakan: skalabilitas horizontal yang masif dengan throughput jutaan pesan per detik.

Jika Kafka harus menjamin urutan di tingkat topik secara global, maka seluruh pesan yang masuk ke topik tersebut harus ditulis ke dalam satu berkas log tunggal secara berurutan. Hal ini memaksa kita untuk hanya menggunakan satu partisi saja untuk topik tersebut. Membatasi topik hanya pada satu partisi akan menghilangkan kemampuan skalabilitas horizontal Kafka:

Bottleneck Broker: Hanya ada satu broker (pemimpin partisi) yang dapat memproses operasi tulis (write) dan baca (read) untuk topik tersebut.
Keterbatasan Skala Consumer: Kita tidak dapat melakukan penskalaan konsumen (consumer scaling) menggunakan grup konsumen (consumer group), karena maksimal hanya satu instans konsumen yang dapat membaca dari partisi tunggal tersebut dalam satu waktu.
Batas Throughput: Throughput sistem akan langsung terbatasi oleh kemampuan I/O disk dan jaringan dari satu mesin broker tunggal.

Oleh karena itu, Kafka memperkenalkan konsep pembagian beban kerja lewat partisi, di mana setiap partisi adalah sebuah berkas log append-only yang independen dan memiliki urutan internalnya sendiri.

Mekanisme Pengurutan Berbasis Kunci (Message Key) #

Agar jaminan urutan di tingkat partisi ini berguna untuk memecahkan masalah bisnis nyata, kita harus memastikan bahwa event-event yang saling berkaitan secara kronologis selalu masuk ke partisi yang sama. Di sinilah peran Message Key menjadi sangat krusial.

Saat mempublikasikan pesan, produser dapat menyertakan sebuah kunci (misalnya, user_id, transaction_id, atau order_id). Secara default, produser Kafka menggunakan algoritma hashing Murmur2 untuk memetakan kunci tersebut menjadi nomor partisi yang spesifik dengan rumus berikut:

$$\text{Partisi} = \text{abs}(\text{MurmurHash2}(\text{Key})) \pmod{\text{Jumlah Partisi}}$$

Dengan mekanisme ini, semua event yang memiliki nilai kunci yang sama dijamin akan selalu dikirim ke partisi yang sama, dan dengan demikian, urutannya akan terjaga secara mutlak di dalam partisi tersebut.

flowchart TD
    subgraph ProducerClient ["Producer (Klien)"]
        direction TB
        E1["Event 1 (Key: User_A)"]
        E2["Event 2 (Key: User_B)"]
        E3["Event 3 (Key: User_A)"]
    end

    subgraph KafkaBroker ["Kafka Broker (Topic: user-actions)"]
        direction TB
        subgraph P0 ["Partition 0"]
            P0_L1["Event 1 (Key: User_A)"]
            P0_L2["Event 3 (Key: User_A)"]
        end
        subgraph P1 ["Partition 1"]
            P1_L1["Event 2 (Key: User_B)"]
        end
    end

    E1 -->|"Hash(User_A) -> P0"| P0
    E2 -->|"Hash(User_B) -> P1"| P1
    E3 -->|"Hash(User_A) -> P0"| P0

    style P0 stroke:#0288d1,stroke-width:2px
    style P1 stroke:#2e7d32,stroke-width:2px

[!WARNING] Dampak Ekspansi Partisi (Partition Expansion): Jika kita menambah jumlah partisi pada topik yang sudah berjalan, hasil perhitungan modulo dari hash kunci akan berubah. Akibatnya, pesan baru dengan kunci yang sama kemungkinan besar akan dipetakan ke partisi yang berbeda dari pesan-pesan sebelumnya. Hal ini akan memecah jaminan urutan data historis untuk kunci tersebut. Jika jaminan urutan sangat krusial, hindari melakukan ekspansi partisi secara dinamis, atau gunakan custom partitioner yang tidak bergantung pada jumlah total partisi secara langsung.

Jaminan Sisi Produser: Mengatasi Out-of-Order Akibat Retry #

Menjamin urutan di broker saja tidak cukup. Masalah pengurutan sering kali justru terjadi dalam perjalanan data dari produser menuju broker. Di lingkungan jaringan terdistribusi yang tidak stabil, kegagalan transmisi data adalah hal yang lumrah terjadi.

Risiko Out-of-Order pada Retry Standar #

Secara default, produser Kafka dikonfigurasi untuk mengirim pesan secara asinkron untuk mencapai throughput maksimal. Produser dapat mengirim beberapa batch request secara bersamaan sebelum menerima pengakuan (acknowledgement) dari broker. Parameter yang mengatur jumlah request paralel ini adalah max.in.flight.requests.per.connection.

Mari kita bedah skenario kegagalan tanpa proteksi pengurutan:

Produser mengirim Batch A (Pesan 1 & 2) dan Batch B (Pesan 3 & 4) secara paralel. max.in.flight.requests.per.connection bernilai 5.
Broker menerima Batch B terlebih dahulu dan sukses menulisnya ke disk.
Transmisi Batch A mengalami kegagalan sementara di jaringan (misalnya, paket hilang).
Broker mengirimkan pesan error ke produser untuk Batch A.
Karena produser dikonfigurasi dengan retries > 0, produser akan mencoba mengirim ulang Batch A.
Broker sukses menerima pengiriman ulang Batch A dan menulisnya ke disk.

Pada akhir skenario ini, pesan di broker akan tertulis dengan urutan: Pesan 3, Pesan 4, Pesan 1, Pesan 2. Urutan kronologis asli telah rusak total!

Solusi Klasik vs Solusi Modern #

Di masa lalu, satu-satunya cara untuk menjamin urutan produser saat menghadapi kegagalan adalah dengan membatasi jumlah request paralel menjadi satu:

# Solusi Klasik (ANTI-PATTERN untuk Throughput Tinggi)
max.in.flight.requests.per.connection=1
retries=2147483647

Dengan konfigurasi ini, produser tidak akan pernah mengirimkan request berikutnya sebelum request sebelumnya mendapatkan ACK sukses. Meskipun ini menjamin urutan 100%, hal ini menurunkan throughput secara drastis karena mengubah proses pengiriman menjadi sepenuhnya sinkron (satu per satu).

Sejak Kafka 0.11, diperkenalkan solusi yang jauh lebih elegan: Idempotent Producer.

# Solusi Modern (Sangat Direkomendasikan)
enable.idempotence=true
max.in.flight.requests.per.connection=5
retries=2147483647

Cara Kerja Idempotent Producer #

Ketika enable.idempotence diset ke true, broker akan mengalokasikan sebuah ID unik yang disebut Producer ID (PID) untuk setiap produser baru selama fase inisialisasi. Selain itu, setiap pesan yang dikirim oleh produser akan diberi Sequence Number yang naik secara monotonik untuk partisi tujuan yang spesifik.

Broker akan melacak Sequence Number terakhir yang berhasil ditulis untuk setiap pasangan PID dan partisi. Ketika broker menerima pesan baru, ia akan memverifikasi Sequence Number-nya:

Jika Incoming SeqNum = Last Written SeqNum + 1, broker menerima pesan tersebut.
Jika Incoming SeqNum <= Last Written SeqNum, broker mendeteksi pesan tersebut sebagai duplikat dan langsung membuangnya tanpa menulis ke log, tetapi tetap mengirimkan ACK sukses ke produser (agar produser tahu pesan sudah aman).
Jika Incoming SeqNum > Last Written SeqNum + 1, broker mendeteksi adanya celah data (out-of-sequence), menolak pesan tersebut dengan error OutOfOrderSequenceException, dan memaksa produser melakukan rekonsiliasi.

Dengan mekanisme penolakan ini, broker menjamin tidak akan pernah ada celah atau pesan yang tertukar urutannya di dalam commit log, bahkan jika produser mengirimkan beberapa in-flight requests secara paralel.

sequenceDiagram
    participant P as "Idempotent Producer"
    participant B as "Kafka Broker"
    
    Note over P,B: Inisialisasi: Producer mendapatkan PID 9999
    
    P->>B: Kirim Pesan A (PID=9999, Seq=0)
    P->>B: Kirim Pesan B (PID=9999, Seq=1)
    
    Note over B: Pesan A berhasil ditulis
    B-->>P: ACK Pesan A (Sukses)
    
    Note over B: Pesan B berhasil ditulis, tapi ACK hilang di jaringan
    B--xP: ACK Pesan B (Gagal Terkirim)
    
    P->>B: Retry Kirim Pesan B (PID=9999, Seq=1)
    Note over B: Broker mendeteksi Seq 1 sudah ada (Duplikat)
    B-->>P: ACK Pesan B (Sukses, Duplikat Diabaikan)
    
    P->>B: Kirim Pesan C (PID=9999, Seq=2)
    B-->>P: ACK Pesan C (Sukses)

Jaminan Sisi Consumer: Mengatasi Masalah Concurrency #

Bahkan jika data kita telah tertulis di broker dengan urutan yang sempurna, semua jaminan tersebut bisa hancur seketika ketika data sampai di sisi aplikasi konsumen (consumer application). Masalah ini hampir selalu bersumber dari satu hal: pemrosesan paralel yang tidak terkelola.

Anti-Pattern: Unpinned Thread Pool #

Secara desain, konsumen Kafka beroperasi dengan model satu utas (single-threaded poll loop). Satu instans KafkaConsumer akan memanggil .poll() untuk mengambil satu batch pesan dari broker, memprosesnya secara berurutan, lalu melakukan komit terhadap offset terakhir.

Namun, dalam dunia nyata, memproses pesan secara berurutan dalam satu thread sering kali terlalu lambat, terutama jika proses tersebut melibatkan operasi I/O yang berat seperti memanggil API eksternal atau menulis ke database SQL. Untuk mempercepat pemrosesan, para developer sering kali tergiur untuk mendistribusikan pesan-pesan dari batch hasil .poll() ke dalam sebuah pool utas (thread pool / ExecutorService) secara asinkron.

Mari kita lihat contoh implementasi Java yang salah ini:

// ANTI-PATTERN: Menghancurkan jaminan pengurutan pesan secara asinkron
public class DangerousConsumer {
    private final ExecutorService threadPool = Executors.newFixedThreadPool(10);
    private final KafkaConsumer<String, String> consumer = new KafkaConsumer<>(configs);

    public void start() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // Mendistribusikan pesan langsung ke thread pool tanpa mempedulikan kunci (key)
                threadPool.submit(() -> {
                    processBusinessLogic(record); // ✗ Urutan rusak di sini!
                });
            }
            // Komit offset secara otomatis atau manual di sini (kritis: offset terkomit sebelum proses selesai!)
        }
    }
}

Mengapa Kode di Atas Sangat Berbahaya? #

Thread Scheduling Non-deterministik: Bayangkan ada dua pesan dari partisi yang sama masuk dalam satu batch: Pesan 1 (Update Nama User menjadi ‘Budi’) dan Pesan 2 (Update Nama User menjadi ‘Joko’). Jika Pesan 1 didelegasikan ke Thread A dan Pesan 2 ke Thread B, tidak ada jaminan Thread A akan dieksekusi lebih dulu oleh OS. Jika Thread B berjalan lebih cepat, nama user di database akan diubah menjadi ‘Joko’ terlebih dahulu, kemudian ditimpa oleh Thread A menjadi ‘Budi’. Data di database kini tidak sinkron dengan urutan event yang sebenarnya terjadi.
Komit Sebelum Selesai: Offset dikomit segera setelah loop selesai membagikan tugas ke thread pool, bukan setelah tugas benar-benar selesai dieksekusi oleh worker thread. Jika salah satu thread mengalami crash saat memproses pesan, pesan tersebut akan hilang selamanya dari perspektif pengolahan karena offsetnya sudah terlanjur dikomit sebagai sukses.

Solusi: Key-Pinned Worker Pool (Thread-Affinity) #

Untuk meningkatkan throughput melalui pemrosesan paralel tanpa merusak jaminan pengurutan, kita harus menerapkan pola Key-Pinned Worker Pool.

Konsep dasarnya sangat sederhana: Kita membagi beban kerja ke beberapa worker thread, tetapi kita harus menjamin bahwa pesan yang memiliki kunci yang sama (atau dari partisi yang sama) harus selalu dialokasikan ke worker thread yang sama secara konsisten.

flowchart TD
    subgraph ConsumerProcess ["Proses Konsumen Kafka"]
        CThread["Consumer Thread (Poll Loop)"]
        
        subgraph KeyPinnedPool ["Key-Pinned Worker Pool"]
            direction TB
            Q0["Worker Queue 0"]
            Q1["Worker Queue 1"]
            Q2["Worker Queue 2"]
            
            T0["Worker Thread 0"]
            T1["Worker Thread 1"]
            T2["Worker Thread 2"]
            
            Q0 --> T0
            Q1 --> T1
            Q2 --> T2
        end

        CThread -->|"Hash(Key_A) % 3 -> Queue 0"| Q0
        CThread -->|"Hash(Key_B) % 3 -> Queue 1"| Q1
        CThread -->|"Hash(Key_A) % 3 -> Queue 0"| Q0
    end

    style KeyPinnedPool stroke:#e5e7eb
    style Q0 stroke:#0288d1,stroke-width:2px
    style Q1 stroke:#2e7d32,stroke-width:2px
    style Q2 stroke:#f57c00,stroke-width:2px

Berikut adalah contoh implementasi Java yang kokoh untuk pola Key-Pinned Worker Pool:

// BENAR: Mengamankan jaminan urutan menggunakan Key-Pinned Worker Pool
public class SafeKeyPinnedConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final List<BlockingQueue<Runnable>> workerQueues;
    private final List<Thread> workerThreads;
    private final int numWorkers;

    public SafeKeyPinnedConsumer(int numWorkers, Properties configs) {
        this.numWorkers = numWorkers;
        this.consumer = new KafkaConsumer<>(configs);
        this.workerQueues = new ArrayList<>(numWorkers);
        this.workerThreads = new ArrayList<>(numWorkers);

        // Inisialisasi antrean dan thread pekerja secara manual
        for (int i = 0; i < numWorkers; i++) {
            BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(1000);
            workerQueues.add(queue);
            
            final int workerId = i;
            Thread thread = new Thread(() -> {
                try {
                    while (!Thread.currentThread().isInterrupted()) {
                        Runnable task = queue.take();
                        task.run();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }, "Worker-Thread-" + workerId);
            
            workerThreads.add(thread);
            thread.start();
        }
    }

    public void start() {
        try {
            consumer.subscribe(Collections.singletonList("user-actions"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // Cari indeks worker thread berdasarkan hash kunci pesan
                    int workerIndex = getWorkerIndex(record.key());
                    BlockingQueue<Runnable> targetQueue = workerQueues.get(workerIndex);

                    // Kirim pekerjaan ke worker thread yang spesifik untuk kunci ini (blocking write)
                    targetQueue.put(() -> {
                        processBusinessLogic(record);
                    });
                }
                // Catatan: Manajemen komit offset yang aman memerlukan pelacakan offset 
                // per thread yang berhasil diselesaikan sebelum melakukan komit ke broker.
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            close();
        }
    }

    private int getWorkerIndex(String key) {
        if (key == null) {
            // Jika tidak ada kunci, kirim ke worker acak atau fallback ke partisi
            return 0;
        }
        // Pastikan nilai indeks selalu bernilai positif
        return Math.abs(key.hashCode()) % numWorkers;
    }

    private void processBusinessLogic(ConsumerRecord<String, String> record) {
        // Logika bisnis kita berjalan aman di sini secara sekuensial per kunci
        System.out.printf("[%s] Memproses kunci: %s, Nilai: %s\n", 
            Thread.currentThread().getName(), record.key(), record.value());
    }

    public void close() {
        for (Thread thread : workerThreads) {
            thread.interrupt();
        }
        consumer.close();
    }
}

Penanganan Kasus Kegagalan Tanpa Merusak Urutan #

Tantangan terbesar berikutnya dalam menjaga urutan data adalah ketika kita menghadapi kesalahan pemrosesan (poison pill / pesan yang rusak). Bagaimana kita menangani pesan yang gagal diproses tanpa merusak jaminan urutan data berikutnya?

Dilema Dead Letter Queue (DLQ) #

Dalam arsitektur pesan standar, pendekatan umum untuk menangani kegagalan adalah dengan mengirimkan pesan yang gagal ke topik khusus bernama Dead Letter Queue (DLQ), kemudian konsumen melanjutkan ke pesan berikutnya.

Namun, dalam sistem yang membutuhkan pengurutan ketat, pendekatan DLQ standar ini adalah kesalahan fatal.

flowchart TD
    subgraph Orig ["Alur Pesan Asli di Broker"]
        P1["Pesan 1 (Update Alamat: Jl. A)"] --> P2["Pesan 2 (Update Alamat: Jl. B)"]
    end
    
    subgraph Fail ["Skenario Kegagalan"]
        direction TB
        Step1["1. Konsumen memproses Pesan 1 (terjadi error database, misal: timeout)"] --> Step2["2. Konsumen mengirim Pesan 1 ke DLQ"]
        Step2 --> Step3["3. Konsumen membaca dan memproses Pesan 2 (Jl. B)<br>Sukses memperbarui alamat di DB menjadi 'Jl. B'"]
        Step3 --> Step4["4. Tim ops mengambil Pesan 1 dari DLQ dan memproses ulang<br>Alamat di DB diperbarui kembali menjadi 'Jl. A'"]
    end

Hasil akhirnya: Data di database salah (kembali ke Jl. A), padahal perubahan terakhir yang sah dari pengguna adalah Jl. B.

Strategi Penanganan Error yang Aman #

Jika kita tidak bisa langsung membuang pesan ke DLQ, apa pilihan alternatif kita?

1. Stop and Retry (Blocking Retry) #

Konsumen berhenti memproses data untuk partisi tersebut dan mencoba kembali secara terus-menerus (retry) hingga pesan tersebut berhasil diproses atau masalah infrastruktur teratasi.

Kelebihan: Urutan data terjamin 100% karena pesan berikutnya tidak akan diproses sebelum pesan yang bermasalah selesai.
Kekurangan: Menyebabkan lag pemrosesan yang parah. Satu pesan yang rusak (poison pill) akan menyandera seluruh partisi dan menghentikan pemrosesan untuk ribuan pengguna lain yang kebetulan berbagi partisi yang sama.

2. Pause & Resume Consumer (Solusi Direkomendasikan) #

Untuk memitigasi dampak dari blocking retry, kita bisa menggunakan API pause() dan resume() dari konsumen Kafka.

Saat terjadi kegagalan pemrosesan pada partisi tertentu:

Kita memanggil consumer.pause(collections.singleton(partition)) untuk menghentikan pengambilan pesan baru khusus untuk partisi tersebut.
Konsumen tetap melanjutkan pemrosesan data untuk partisi-partisi lain yang sehat tanpa terganggu.
Kita mengirim pesan yang gagal ke mekanisme retri internal (misal, menyimpannya di database lokal atau memori sementara dengan masa tunggu tertentu).
Setelah masa tunggu atau perbaikan database selesai, kita memproses ulang pesan tersebut.
Jika sukses, kita memanggil consumer.resume(collections.singleton(partition)) untuk mulai mengambil data lagi dari partisi tersebut.

Dengan metode ini, lag pemrosesan hanya akan berdampak pada pengguna-pengguna yang datanya berada di partisi yang bermasalah, sedangkan partisi lainnya tetap dapat memproses data dengan throughput maksimal.

Ringkasan #

Jaminan Terbatas: Apache Kafka hanya menjamin urutan pesan pada tingkat partisi (partition-level), bukan di tingkat topik secara global (topic-level).

Peran Kunci (Message Key): Gunakan kunci pesan secara konsisten untuk memastikan event yang saling bergantung secara kronologis masuk ke partisi yang sama melalui algoritma hash Murmur2.

Idempotent Producer: Selalu aktifkan enable.idempotence=true di sisi produser untuk mencegah rusaknya urutan data akibat skenario pengiriman ulang (retry) di jaringan.

Bahaya Concurrency Consumer: Hindari mendistribusikan batch hasil .poll() langsung ke thread pool standar. Gunakan pola Key-Pinned Worker Pool untuk mempertahankan urutan di sisi aplikasi.

Dilema DLQ: Mengirim pesan gagal ke DLQ secara instan akan merusak urutan logis data. Gunakan strategi blocking retry atau manfaatkan fitur Pause & Resume untuk isolasi error yang aman.

← Sebelumnya: Partition Strategy Berikutnya: Acknowledgement →