Auto vs Manual Commit #

Ketika aplikasi konsumen kita membaca pesan dari Apache Kafka, broker tidak melacak pesan mana saja yang telah sukses diproses oleh konsumen. Broker hanya mencatat penanda posisi baca terakhir dari kelompok konsumen kita untuk setiap partisi. Penanda posisi ini disimpan di dalam topik internal khusus bernama __consumer_offsets dan dikenal sebagai Offset Komit (Committed Offset). Setiap kali konsumen kita dinyalakan ulang atau terjadi proses pembagian partisi ulang (rebalance), konsumen baru akan membaca posisi awal pemrosesan dari offset komit terakhir ini. Oleh karena itu, ketepatan waktu dan keamanan saat mengirimkan data offset ke broker adalah faktor paling kritis yang menentukan keandalan transmisi data kita. Kita dihadapkan pada pilihan mendasar: menggunakan mekanisme komit otomatis (Auto Commit) yang nyaman namun berbahaya, atau mengendalikan penyimpanan offset secara manual (Manual Commit) untuk menjamin integritas data bisnis kita.

Mekanisme Auto-Commit (Komit Otomatis) #

Secara default, klien konsumen Kafka dikonfigurasi untuk menggunakan mekanisme komit otomatis dengan properti enable.auto.commit=true.

Cara Kerja Auto-Commit #

Ketika fitur ini aktif, konsumen akan secara berkala mengumumkan offset tertinggi yang diperolehnya dari panggilan .poll() ke broker coordinator. Interval waktu berkala ini dikontrol oleh properti auto.commit.interval.ms yang memiliki nilai default sebesar 5.000 ms (5 detik).

Mekanisme ini berjalan di dalam utas poll loop utama secara pasif:

Aplikasi memanggil consumer.poll(). Klien konsumen secara internal memeriksa apakah waktu sejak komit terakhir telah melampaui batas 5 detik.
Jika batas 5 detik telah terlewati, konsumen akan melampirkan permintaan komit offset terakhir ke dalam permintaan I/O jaringan berikutnya (misalnya saat mengirimkan heartbeat atau melakukan fetch request baru).
Broker memperbarui offset di __consumer_offsets, dan siklus 5 detik dimulai kembali dari nol.

Mekanisme ini sangat populer karena kesederhanaannya — tim pengembang tidak perlu menulis satu baris kode pun untuk mengelola offset, karena Kafka client menangani semuanya di latar belakang. Namun, kenyamanan ini menyimpan risiko kegagalan fatal yang dapat merusak kualitas data kita di lingkungan produksi.

Bahaya Maut Menggunakan Auto-Commit di Produksi #

Meskipun mempermudah kode penulisan, membiarkan properti enable.auto.commit=true aktif di lingkungan produksi yang memiliki lalu lintas data bernilai tinggi sangat tidak direkomendasikan. Mekanisme berbasis waktu yang pasif ini dapat memicu dua skenario bencana data: Kehilangan Data (Data Loss) dan Duplikasi Data (Data Duplication).

Skenario 1: Kehilangan Data (Data Loss) #

Bayangkan skenario berikut terjadi pada aplikasi konsumen kita:

Konsumen memanggil .poll() dan menerima 100 pesan (Offset 100 hingga 199).
Pesan-pesan ini disimpan sementara di antrean memori heap aplikasi.
Utas utama mulai memproses pesan-pesan ini satu per satu. Logika bisnis di setiap pesan memakan waktu 100 milidetik (total butuh 10 detik untuk menyelesaikan seluruh batch).
Kejadian Kritis: Tepat pada detik ke-5 pemrosesan (ketika aplikasi baru sukses memproses 50 pesan hingga Offset 149), utas utama memanggil .poll() berikutnya untuk menjaga keaktifan loop. Karena waktu 5 detik telah habis, klien konsumen secara otomatis mengirimkan komit offset ke broker untuk Offset 199 (karena Offset 199 adalah offset tertinggi yang sempat ditarik di langkah 1). Broker mencatat Offset 199 sebagai posisi komit sukses.
Pada detik ke-6, aplikasi kita mendadak mengalami crash (misal terkena OutOfMemory, mati listrik, atau di-restart oleh Kubernetes scheduler). Pemrosesan pesan Offset 150 hingga 199 gagal total dan belum dieksekusi.
Saat kontainer aplikasi pulih dan konsumen dinyalakan kembali, konsumen meminta offset awal ke broker. Broker mengembalikan angka Offset 199.
Hasil Akhir: Konsumen mulai membaca dari Offset 200. Pesan dari Offset 150 hingga 199 hilang secara permanen dari siklus pemrosesan aplikasi kita tanpa ada error trace yang tercatat.

Skenario 2: Duplikasi Data (Data Duplication) #

Kebalikan dari skenario di atas, jika aplikasi kita sukses memproses 90 pesan (hingga Offset 189) dalam waktu 4 detik, lalu mengalami crash sebelum interval 5 detik habis:

Broker belum sempat menerima pembaruan offset otomatis, sehingga offset komit terakhir di broker masih tertinggal di angka Offset 99.
Saat konsumen baru dinyalakan, ia dipaksa membaca ulang data dari Offset 100.
Hasil Akhir: Pesan Offset 100 hingga 189 diproses untuk kedua kalinya, memicu duplikasi data di database hilir kita.

Alternatif Terbaik: Manual Commit (Komit Manual) #

Untuk menjamin tingkat keamanan data maksimum, kita harus mematikan fitur komit otomatis dan mengendalikan sepenuhnya kapan offset harus dikomit secara manual:

enable.auto.commit=false

Dengan mematikan auto-commit, kita dapat menerapkan aturan bisnis yang ketat: Komit offset hanya boleh dikirim ke broker jika dan hanya jika seluruh rangkaian pesan di dalam batch poll tersebut telah selesai diproses dan disimpan dengan sukses ke penyimpanan database utama kita.

Untuk memicu komit secara manual, Java client SDK menyediakan dua metode utama: commitSync() dan commitAsync().

Membedah commitSync() vs commitAsync() #

Memilih di antara kedua metode manual ini merupakan keputusan arsitektural yang melibatkan trade-off antara kecepatan throughput dengan keandalan pemulihan kesalahan.

1. `commitSync()` (Komit Manual Sinkron) #

Metode ini bertindak secara memblokir thread (blocking). Ketika dipanggil, thread utama konsumen akan berhenti berputar dan menunggu hingga broker coordinator selesai menulis offset ke topik __consumer_offsets dan mengembalikan paket respons konfirmasi sukses.

Kelebihan: Keamanan tinggi. Jika terjadi kegagalan transient (seperti jaringan sibuk atau pemilihan leader baru), commitSync() akan secara otomatis melakukan upaya coba ulang (retry) hingga batas timeout tercapai sebelum akhirnya melempar exception ke aplikasi.
Kekurangan: Latensi transmisi tinggi. Menunggu ACK jaringan di setiap perputaran loop membatasi throughput maksimal konsumen kita.

2. `commitAsync()` (Komit Manual Asinkron) #

Metode ini bertindak secara non-blocking. Utas konsumen hanya mengirimkan permintaan komit ke jaringan, lalu langsung melanjutkan perputaran loop untuk memproses data berikutnya tanpa pernah menunggu respons dari broker.

Kelebihan: Throughput sangat tinggi dan latensi minimal.
Kekurangan: Tidak melakukan retry otomatis jika terjadi kegagalan. Mengapa? Karena retry asinkron dapat memicu masalah Commit Race Condition.

Masalah Commit Race Condition (Balapan Komit) #

Bayangkan jika konsumen kita mengirimkan dua permintaan komit asinkron secara berurutan:

Konsumen memanggil commitAsync(Offset 100). Permintaan ini terhambat di jaringan sementara.
Konsumen memproses batch berikutnya dan memanggil commitAsync(Offset 200). Permintaan ini sukses sampai di broker lebih cepat. Broker mencatat offset terakhir adalah 200.
Permintaan pertama commitAsync(Offset 100) akhirnya berhasil lolos dari kemacetan jaringan dan sampai di broker.
Hasil Akhir: Jika commitAsync() melakukan retry otomatis untuk kegagalan lama, ia akan menimpa offset 200 kembali menjadi 100 secara keliru. Oleh karena itu, commitAsync() didesain untuk tidak pernah melakukan retry jika terjadi kegagalan.

Pola Kombinasi Terbaik (The Standard Commit Pattern) #

Untuk mendapatkan keunggulan kecepatan commitAsync() sekaligus keandalan commitSync(), standar industri menyarankan penggunaan kombinasi pola berikut:

Gunakan commitAsync() di dalam perulangan loop utama untuk performa throughput maksimal tanpa memblokir thread pemroses.
Bungkus seluruh blok loop di dalam penanganan try-catch-finally.
Di dalam blok finally (yang dipicu saat aplikasi menerima sinyal shutdown), panggil commitSync() satu kali secara paksa untuk memastikan offset terakhir yang tersisa sebelum aplikasi mati benar-benar tersimpan aman di broker.

Diagram Mermaid: Alur Auto Commit vs Manual Commit #

Diagram urutan berikut membandingkan bagaimana status offset dikelola antara model Auto Commit yang berisiko data loss dengan model Manual Commit yang aman:

sequenceDiagram
    autonumber
    participant App as Utas Aplikasi
    participant Cons as Kafka Consumer Client
    participant Broker as Kafka Broker Coordinator
    
    Note over App, Broker: "Skenario 1: Bahaya Auto-Commit (enable.auto.commit=true)"
    App->>Cons: poll(100ms)
    Cons->>Broker: Ambil Data (Offset 100-110)
    Broker-->>Cons: Return 10 records
    Cons-->>App: Return records
    App->>App: "Mulai Proses Bisnis (Butuh waktu 6 detik)"
    Note over Cons, Broker: "Detik ke-5: Auto-Commit dipicu otomatis saat poll berikutnya"
    App->>Cons: poll(100ms)
    Cons->>Broker: CommitOffsetRequest (Offset 110)
    Broker-->>Cons: Commit ACK
    Note over App: "Detik ke-6: Aplikasi Crash / OOM saat memproses data!"
    Note over App: "Saat restart, konsumen baru membaca dari Offset 110 (Data 100-109 HILANG!)"
    
    Note over App, Broker: "Skenario 2: Kontrol Manual Commit (enable.auto.commit=false)"
    App->>Cons: poll(100ms)
    Cons->>Broker: Ambil Data (Offset 200-210)
    Broker-->>Cons: Return 10 records
    Cons-->>App: Return records
    App->>App: "Proses Bisnis Sukses (Tulis ke DB)"
    App->>Cons: commitSync()
    Cons->>Broker: CommitOffsetRequest (Offset 210)
    Broker-->>Cons: Commit ACK (Offset tersimpan aman)

Implementasi Java: Penyetelan Pola Komit Manual Terbaik #

Berikut adalah perbandingan kode Java antara penggunaan auto-commit yang rentan kehilangan data (anti-pattern) dengan implementasi kombinasi manual commit (commitAsync + commitSync) yang direkomendasikan untuk stabilitas produksi:

// ANTI-PATTERN: Mengandalkan auto-commit berbasis waktu untuk data transaksi keuangan
public class UnsafePaymentConsumer {
    public void startConsuming(Properties props) {
        // ✗ Sangat berbahaya: Menggunakan auto-commit default untuk transaksi penting
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("payments"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    // Jika proses ini crash di tengah, data setelah detik ke-5 akan terkomit otomatis
                    // dan memicu kehilangan data permanen
                    executePayment(record.value());
                }
            }
        }
    }
    private void executePayment(String val) {}
}

// BENAR: Menggunakan kombinasi manual commitAsync() dan commitSync() secara aman
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class SafeTransactionalConsumer {
    public void consumeSecurely(Properties props) {
        // ✓ BENAR: Matikan kompresi otomatis untuk mengambil alih kendali penuh
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("payments"));

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    processPayment(record.value());
                }
                
                // ✓ BENAR: Gunakan commitAsync() untuk menjaga throughput tetap tinggi di loop utama
                // Kita melewatkan callback kosong atau logger sederhana untuk memantau error jaringan
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(java.util.Map<org.apache.kafka.common.TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            // Catat log jika terjadi kegagalan komit asinkron
                            System.err.println("Gagal mengomit offset secara asinkron: " + exception.getMessage());
                        }
                    }
                });
            }
        } catch (WakeupException e) {
            System.out.println("Konsumen dibangunkan untuk shutdown...");
        } catch (Exception e) {
            System.err.println("Error fatal pada proses konsumsi: " + e.getMessage());
        } finally {
            try {
                // ✓ BENAR: Gunakan commitSync() secara memblokir di blok finally
                // Ini memastikan offset terakhir sebelum aplikasi mati benar-benar tertulis di broker
                System.out.println("Melakukan komit sinkron terakhir sebelum keluar...");
                consumer.commitSync();
            } finally {
                consumer.close();
                System.out.println("Konsumen sukses ditutup secara graceful.");
            }
        }
    }

    private void processPayment(String payload) {
        System.out.println("Sukses memproses pembayaran: " + payload);
    }
}

Memahami CommitFailedException saat Rebalance #

Salah satu tantangan terbesar saat menggunakan komit manual di lingkungan produksi adalah penanganan CommitFailedException. Pengecualian (exception) ini dilempar secara sinkron oleh commitSync() (atau dilaporkan melalui callback asinkron pada commitAsync()) jika broker coordinator mendeteksi bahwa partisi yang sedang coba dikomit oleh konsumen kita telah dialokasikan ke konsumen lain.

Mengapa Ini Terjadi? #

Skenario ini paling sering dipicu oleh terlampauinya batas waktu pemrosesan max.poll.interval.ms.

Konsumen memanggil .poll(), mendapatkan batch data, lalu utas utama memproses data secara lambat (melebihi batas waktu default 5 menit).
Broker Coordinator menyimpulkan bahwa konsumen tersebut telah mati (stuck/livelock). Coordinator mengeluarkan konsumen dari kelompok dan memicu proses Rebalance.
Partisi-partisi konsumen tersebut dialokasikan ke konsumen baru yang masih sehat. Konsumen baru mulai membaca data dari offset terakhir yang tercatat di broker.
Konsumen lama akhirnya selesai memproses data lambatnya dan memanggil consumer.commitSync().
Broker Coordinator menolak permintaan tersebut dan melempar CommitFailedException karena status kepemilikan partisi telah berpindah tangan.

Cara Penanganan yang Benar #

Kita tidak boleh mencoba mengulang (retry) komit yang gagal akibat CommitFailedException. Kita harus menangkap exception ini, mencatat log audit, membuang status pemrosesan lokal, dan membiarkan konsumen baru mengambil alih. Upaya memaksa menulis offset dari konsumen yang telah dikeluarkan hanya akan merusak data offset milik konsumen baru.

Komit Granular: Menentukan Offset Spesifik Per Partisi #

Secara default, pemanggilan commitSync() atau commitAsync() tanpa argumen akan mengomit semua offset terakhir dari seluruh partisi yang dikembalikan oleh panggilan .poll() sebelumnya. Pendekatan ini terkadang dinilai kurang presisi untuk sistem dengan throughput sangat padat.

Untuk memberikan kontrol penuh, klien Java menyediakan fungsi komit yang menerima parameter spesifik:

// Melakukan komit offset secara granular untuk partisi tertentu saja
consumer.commitSync(Map<TopicPartition, OffsetAndMetadata> offsets);

Keunggulan Komit Granular #

Dengan memetakan offset secara eksplisit per partisi, kita dapat melakukan komit secara bertahap (mid-batch commits). Sebagai contoh, jika sebuah panggilan .poll() mengembalikan record dari 3 partisi (Partisi 0, Partisi 1, dan Partisi 2), aplikasi kita dapat:

Memproses seluruh record khusus untuk Partisi 0.
Segera melakukan komit offset hanya untuk Partisi 0.
Melanjutkan proses untuk Partisi 1 dan seterusnya.

Hal ini secara drastis meminimalkan jumlah pesan yang harus diproses ulang (duplikasi) jika aplikasi kita mengalami crash di tengah-tengah pemrosesan batch yang besar.

Berikut contoh implementasi komit granular di Java:

// ✓ BENAR: Mengomit offset secara granular per partisi untuk meminimalkan duplikasi data
public void consumeGranularly(KafkaConsumer<String, String> consumer) {
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
            
            for (ConsumerRecord<String, String> record : partitionRecords) {
                processRecord(record);
            }
            
            // Dapatkan offset dari record terakhir di partisi ini, ditambah 1 (posisi baca berikutnya)
            long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
            
            // Buat map komit untuk partisi spesifik ini
            Map<TopicPartition, OffsetAndMetadata> commitMap = Collections.singletonMap(
                partition, 
                new OffsetAndMetadata(lastOffset + 1)
            );
            
            // ✓ Komit offset hanya untuk partisi yang telah selesai diproses sepenuhnya
            consumer.commitSync(commitMap);
        }
    }
}

Ringkasan #

Offset Commit: Mekanisme pencatatan posisi baca terakhir kelompok konsumen di broker Kafka yang disimpan pada topik internal __consumer_offsets.

Auto-Commit Danger: Fitur auto-commit berbasis waktu (5s) rentan memicu bencana kehilangan data jika aplikasi crash di tengah-tengah pemrosesan batch.

Manual Control: Mematikan auto-commit (enable.auto.commit=false) adalah standar wajib di produksi untuk menjamin pemrosesan data presisi.

commitSync: Operasi komit blocking yang menjamin keamanan penulisan offset dengan melakukan retry otomatis jika terjadi gangguan jaringan sementara.

commitAsync: Operasi komit non-blocking yang mempertahankan kecepatan throughput maksimal dengan mengeliminasi waktu tunggu respons jaringan.

Combined Pattern: Pola desain terbaik menggunakan commitAsync() pada perulangan loop utama dan menyegelnya dengan commitSync() saat shutdown.

← Sebelumnya: Poll Loop Berikutnya: Offset Management →