Offset Management #

Ketika kita merancang sistem pemrosesan aliran data terdistribusi yang toleran terhadap kegagalan (fault-tolerant), salah satu tantangan terbesar adalah menjaga sinkronisasi posisi baca. Seperti yang kita bahas pada bab sebelumnya, menyimpan offset komit di dalam penyimpanan internal Kafka (__consumer_offsets) — baik menggunakan auto-commit maupun manual commit — tidak pernah bersifat atomik dengan penulisan transaksi database bisnis kita. Hal ini dikarenakan menulis ke Kafka broker dan menulis ke database RDBMS (seperti PostgreSQL atau MySQL) merupakan dua jaringan I/O terpisah yang tidak dibungkus dalam satu transaksi global. Jika aplikasi crash di sela-sela kedua penulisan tersebut, kita tidak bisa menghindari risiko duplikasi data. Untuk mencapai jaminan pemrosesan Exactly-Once sejati tanpa overhead transaksi Kafka internal, arsitektur microservices tingkat lanjut menerapkan strategi External Offset Storage (Penyimpanan Offset Eksternal) dengan melakukan manajemen offset secara mandiri menggunakan bantuan API consumer.seek().

Masalah Sistem Terdistribusi Non-Atomik #

Secara default, alur kerja konsumen umum yang menulis data ke database adalah sebagai berikut:

  1. Konsumen memanggil .poll() dan memperoleh pesan dengan Offset 500.
  2. Konsumen memproses data dan menulis hasilnya ke tabel database bisnis orders (sukses).
  3. Konsumen mengirimkan komit manual commitSync(Offset 500) ke broker Kafka (gagal karena jaringan putus).
  4. Masalah: Data bisnis sudah tersimpan di database, namun Kafka coordinator menganggap offset terakhir masih 499. Saat konsumen pulih, ia akan memproses ulang pesan 500, memicu duplikasi data di database.

Kita tidak bisa membungkus operasi penyimpanan data SQL di PostgreSQL dengan operasi komit Kafka broker ke dalam satu blok transaksi ACID yang sama. Keduanya merupakan sistem penyimpanan terpisah (non-atomic distributed writes).

Solusi: Menyimpan Offset di Database Eksternal (RDBMS) #

Cara paling elegan untuk menyelesaikan masalah non-atomik ini adalah dengan memindahkan lokasi penyimpanan offset komit dari topik internal Kafka __consumer_offsets ke dalam tabel khusus di dalam database RDBMS bisnis kita sendiri.

Konsep Transaksi Database Lokal (ACID) #

Dengan menyimpan offset di database bisnis yang sama dengan data bisnis kita, kita dapat membungkus operasi penyimpanan data bisnis dan pembaruan nomor offset Kafka ke dalam satu transaksi lokal database yang sama (BEGIN TRANSACTION … COMMIT):

BEGIN TRANSACTION;

-- 1. Tulis data transaksi bisnis
INSERT INTO orders (order_id, customer_id, amount) VALUES ('TX-1001', 'CUST-99', 500000.0);

-- 2. Update offset Kafka untuk partisi terkait
UPDATE kafka_offsets 
SET last_offset = 501 
WHERE topic = 'orders-topic' AND partition_id = 2;

COMMIT;

Mengapa Pendekatan Ini 100% Aman? #

Di dalam RDBMS terstruktur, transaksi dijamin mematuhi prinsip ACID (Atomicity, Consistency, Isolation, Durability).

  • Jika koneksi database putus di tengah jalan, atau salah satu dari kedua perintah SQL di atas gagal, database akan melakukan rollback secara utuh untuk kedua baris perintah tersebut. Tidak akan pernah ada kondisi di mana data bisnis tersimpan namun offset tidak bertambah, atau sebaliknya.
  • Sisi Kafka broker tidak pernah menerima komit offset sama sekali (kita mematikan auto-commit dan tidak pernah memanggil commitSync atau commitAsync pada klien konsumen kita). Kafka coordinator akan menganggap offset grup kita tetap di angka lama, namun kita mengabaikan status offset di broker tersebut dan menggunakan status offset lokal di database kita sebagai kebenaran mutlak.

Alur Kerja Transaksional ke Database MySQL #

Bagan alir berikut memvisualisasikan bagaimana siklus pembacaan data, eksekusi transaksi database ACID yang menggabungkan penulisan data bisnis dengan offset Kafka, hingga penanganan rollback jika terjadi crash:

flowchart TD
    Start["1. Aplikasi Memanggil poll()"] --> Cons["Konsumen Menerima Batch Pesan (Offset 500-505)"]
    Cons --> DB_Tx["2. Buka Transaksi Database Lokal (DB.beginTransaction)"]
    
    subgraph Database ["Satu Transaksi Atomik Database MySQL"]
        direction TB
        DB_Write["3. Tulis Data Bisnis (Insert into orders)"] --> DB_Offset["4. Tulis Offset Kafka (Update kafka_offsets set offset = 506 where partition = 0)"]
    end
    
    DB_Tx --> DB_Write
    DB_Offset --> CommitCheck{"Apakah semua tulis database sukses?"}
    
    CommitCheck -- "Ya" --> DB_Commit["5. Komit Transaksi (DB.commit)"]
    DB_Commit --> ProcessNext["6. Siap poll batch data berikutnya"]
    
    CommitCheck -- "Tidak (Crash / Error)" --> DB_Rollback["5. Rollback Transaksi (DB.rollback)"]
    DB_Rollback --> SeekBack["6. Cari Ulang Offset Terakhir di Database & Panggil consumer.seek(partition, 500)"]
    SeekBack --> ProcessNext
    
    style Start stroke:#e5e7eb
    style DB_Tx stroke:#f57c00,stroke-width:2px
    style Database stroke:#e5e7eb
    style DB_Write stroke:#0288d1,stroke-width:2px
    style DB_Offset stroke:#0288d1,stroke-width:2px
    style DB_Commit stroke:#2e7d32,stroke-width:2px
    style DB_Rollback stroke:#c62828,stroke-width:2px

Mekanisme Seeking Offset secara Dinamis #

Untuk menerapkan strategi ini, klien konsumen kita harus mampu melompati posisi baca offset bawaan broker dan mengarahkan pointer baca secara dinamis ke offset yang tercatat di database kita. Tugas ini dijalankan menggunakan API consumer.seek().

Cara Menggunakan consumer.seek() #

Metode seek memungkinkan kita memindahkan pointer baca partisi secara manual:

// Mengarahkan konsumen untuk membaca Partisi 2 mulai dari Offset 501
consumer.seek(new TopicPartition("orders-topic", 2), 501);

Integrasi dengan ConsumerRebalanceListener #

Karena kepemilikan partisi dapat berpindah-pindah antar konsumen akibat proses rebalance, kita tidak bisa memanggil seek hanya saat inisialisasi awal aplikasi. Kita wajib mendaftarkan objek ConsumerRebalanceListener saat memanggil fungsi .subscribe().

Listener ini memiliki dua metode utama yang dicegat saat rebalance terjadi:

  • onPartitionsRevoked: Dipanggil sebelum partisi kita dicopot. Di sini kita harus memastikan seluruh tugas di partisi tersebut selesai ditulis dan dikomit ke database.
  • onPartitionsAssigned: Dipanggil setelah partisi baru dialokasikan ke konsumen kita. Di sinilah kita harus melakukan query SQL ke database bisnis untuk mencari offset terakhir yang sukses dikomit untuk masing-masing partisi baru tersebut, lalu memanggil consumer.seek() untuk masing-masing partisi sebelum jajak pendapat .poll() pertama dimulai.

Implementasi Java: Transaksi Bisnis & Offset Terpadu #

Berikut adalah perbandingan kode Java antara penulisan offset standard yang tidak aman (anti-pattern) dengan penulisan transaksional database yang aman menggunakan JDBC dan ConsumerRebalanceListener kustom:

// ANTI-PATTERN: Menulis ke database dan melakukan komit offset ke Kafka secara terpisah
// Sangat rawan duplikasi data jika koneksi internet terputus sebelum komit offset berhasil dikirim
public class UnsafeDatabaseWriter {
    public void consume(KafkaConsumer<String, String> consumer, Connection dbConnection) {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    // 1. Tulis data bisnis ke database
                    saveOrderToDb(dbConnection, record.value());
                    
                    // 2. Komit ke Kafka broker secara terpisah (Non-Atomik)
                    consumer.commitSync(); // ✗ Risiko crash di sini memicu duplikasi data
                } catch (Exception e) {
                    System.err.println("Gagal menulis data: " + e.getMessage());
                }
            }
        }
    }
    private void saveOrderToDb(Connection conn, String data) {}
}

// BENAR: Menggunakan Transaksi Database Lokal ACID untuk menyatukan Data Bisnis dan Offset Kafka
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.Duration;
import java.util.Collection;
import java.util.Collections;
import java.util.Properties;

public class TransactionalOffsetConsumer {
    private final Connection dbConnection;
    private final KafkaConsumer<String, String> consumer;
    private final String topicName = "orders-topic";

    public TransactionalOffsetConsumer(Properties kafkaProps, Connection dbConnection) {
        this.dbConnection = dbConnection;
        // ✓ BENAR: Matikan auto-commit Kafka secara mutlak
        kafkaProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        this.consumer = new KafkaConsumer<>(kafkaProps);
    }

    public void start() {
        // ✓ BENAR: Daftarkan Rebalance Listener kustom saat berlangganan topik
        consumer.subscribe(Collections.singletonList(topicName), new DatabaseOffsetRebalanceListener());

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (TopicPartition partition : records.partitions()) {
                    for (ConsumerRecord<String, String> record : records.records(partition)) {
                        // ✓ Jalankan penulisan dalam satu transaksi database tunggal
                        processRecordWithinDatabaseTransaction(record);
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }

    private void processRecordWithinDatabaseTransaction(ConsumerRecord<String, String> record) {
        try {
            // ✓ BENAR: Matikan auto-commit JDBC untuk memulai transaksi database lokal
            dbConnection.setAutoCommit(false);

            // 1. Tulis data bisnis
            String insertOrderSql = "INSERT INTO orders (order_id, val) VALUES (?, ?)";
            try (PreparedStatement orderStmt = dbConnection.prepareStatement(insertOrderSql)) {
                orderStmt.setString(1, record.key());
                orderStmt.setString(2, record.value());
                orderStmt.executeUpdate();
            }

            // 2. Tulis offset Kafka ke tabel khusus di database yang sama
            String updateOffsetSql = "INSERT INTO kafka_offsets (topic, partition_id, last_offset) " +
                                     "VALUES (?, ?, ?) " +
                                     "ON CONFLICT (topic, partition_id) " +
                                     "DO UPDATE SET last_offset = EXCLUDED.last_offset";
            try (PreparedStatement offsetStmt = dbConnection.prepareStatement(updateOffsetSql)) {
                offsetStmt.setString(1, record.topic());
                offsetStmt.setInt(2, record.partition());
                // Simpan offset record saat ini + 1 (menunjuk ke posisi baca berikutnya)
                offsetStmt.setLong(3, record.offset() + 1);
                offsetStmt.executeUpdate();
            }

            // ✓ BENAR: Komit transaksi database lokal secara utuh (Atomik)
            dbConnection.commit();
        } catch (SQLException e) {
            try {
                // Batalkan seluruh operasi jika salah satu perintah SQL gagal
                dbConnection.rollback();
            } catch (SQLException ex) {
                System.err.println("Gagal melakukan rollback: " + ex.getMessage());
            }
            System.err.println("Transaksi gagal, melakukan rollback: " + e.getMessage());
        }
    }

    // Listener Kustom untuk mengarahkan posisi baca konsumen saat terjadi Rebalance
    private class DatabaseOffsetRebalanceListener implements ConsumerRebalanceListener {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // Sebelum partisi dicopet, pastikan seluruh data telah dikomit ke database
            try {
                dbConnection.commit();
            } catch (SQLException e) {
                System.err.println("Gagal komit database saat rebalance: " + e.getMessage());
            }
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // Setelah partisi baru dialokasikan, cari offset terakhir dari database
            for (TopicPartition partition : partitions) {
                long lastCommittedOffset = getOffsetFromDatabase(partition);
                
                if (lastCommittedOffset != -1) {
                    // ✓ BENAR: Paksa konsumen memindahkan pointer baca ke offset dari database
                    consumer.seek(partition, lastCommittedOffset);
                } else {
                    // Jika data belum ada di database, gunakan default seek to beginning / end
                    consumer.seekToBeginning(Collections.singletonList(partition));
                }
            }
        }

        private long getOffsetFromDatabase(TopicPartition partition) {
            String sql = "SELECT last_offset FROM kafka_offsets WHERE topic = ? AND partition_id = ?";
            try (PreparedStatement stmt = dbConnection.prepareStatement(sql)) {
                stmt.setString(1, partition.topic());
                stmt.setInt(2, partition.partition());
                try (ResultSet rs = stmt.executeQuery()) {
                    if (rs.next()) {
                        return rs.getLong("last_offset");
                    }
                }
            } catch (SQLException e) {
                System.err.println("Gagal membaca offset dari database: " + e.getMessage());
            }
            return -1; // Fallback jika data kosong
        }
    }
}

Kelemahan dan Kompleksitas Pola Eksternal #

Meskipun strategi External Offset Storage memberikan jaminan Exactly-Once transaksional yang mutlak, pola ini tidak selalu menjadi pilihan utama karena membawa beberapa konsekuensi teknis:

1. Beban Database yang Meningkat #

Menyimpan offset ke dalam RDBMS berarti kita menambahkan query tulis (insert/update) untuk setiap batch atau setiap record pesan. Pada throughput tinggi (misalnya puluhan ribu pesan per detik), ini dapat memicu kemacetan I/O (write bottlenecks) pada database RDBMS kita.

2. Hilangnya Monitoring Lag Bawaan (Consumer Lag Metric) #

Alat bantu monitoring standar bawaan Kafka (seperti perintah kafka-consumer-groups.sh atau Prometheus exporter metrics) mengukur lag konsumen dengan membandingkan offset terakhir di broker dengan offset komit di topik __consumer_offsets.

  • Karena kita mematikan auto-commit dan tidak pernah mengirimkan komit ke broker, nilai committed offset di broker akan tetap di angka awal atau nol.
  • Indikator lag bawaan akan mendeteksi lag yang sangat tinggi (false alarms), meskipun aplikasi konsumen kita sebenarnya sedang berjalan normal membaca offset dari database.
  • Solusi: Kita harus membuat visualisasi metrik lag kustom dengan membuat script monitoring yang secara periodik membandingkan offset high watermark topik Kafka dengan nilai offset terakhir yang tersimpan di tabel database kafka_offsets kita.

Ringkasan #

  • Non-Atomic Distributed Writes: Penulisan offset ke broker Kafka dan penulisan transaksi bisnis ke database RDBMS lokal tidak dapat dilakukan secara atomik secara bawaan.
  • External Offset Storage: Solusi arsitektur memindahkan penyimpanan offset komit dari topik internal __consumer_offsets ke tabel database RDBMS kita.
  • ACID Transaction: Menyatukan operasi modifikasi data bisnis dan pembaruan nomor offset Kafka ke dalam satu transaksi lokal database database.
  • seek() API: Fungsi klien konsumen Kafka untuk melompati offset broker dan mengarahkan pointer pembacaan secara dinamis ke offset spesifik.
  • Rebalance Listener: Komponen interseptor wajib untuk melakukan seeking offset secara dinamis dari database sesaat setelah proses rebalance partisi selesai.
  • Monitoring Challenge: Penggunaan pola eksternal menyebabkan indikator lag bawaan Kafka tidak akurat, mewajibkan pembuatan dashboard metrik lag kustom.

← Sebelumnya: Auto vs Manual Commit   Berikutnya: Consumer Group →

About | Author | Content Scope | Editorial Policy | Privacy Policy | Disclaimer | Contact