Belajar Apache Kafka - Consumers: Membaca Data dari Kafka

#Pendahuluan

Jika producer adalah sisi tulis, consumer adalah sisi baca — dan di sinilah banyak kesalahan umum terjadi: consumer yang membaca offset salah, commit yang terjadi sebelum data selesai diproses, atau rebalance storm yang membuat cluster tidak stabil.

Episode 6 ini akan membawa kalian menguasai consumer: dari subscription dan polling, lifecycle consumer, mekanisme consumer group dan rebalancing dengan berbagai strategi assignment, hingga pengelolaan offset yang benar. Kalian juga akan belajar tuning parameter seperti fetch.min.bytes, max.poll.records, dan session.timeout.ms.

Pola consumer yang benar adalah kunci sistem yang andal dan mudah di-scale. Mari kita bedah satu per satu.

#Consumer Fundamentals

#Konfigurasi dan Subscription

Consumer membutuhkan bootstrap.servers, deserializer, dan group.id untuk membentuk group. Subscription dilakukan dengan subscribe() ke satu atau beberapa topic:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "order-processor");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 
try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("orders"));
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> r : records) {
            System.out.println(r.key() + " -> " + r.value());
        }
    }
}

#Polling dan ConsumerRecord

poll() adalah jantung consumer: memblokir maksimal sesuai timeout untuk mengambil batch record, sekaligus mengirim heartbeat ke koordinator group. Setiap ConsumerRecord membawa topic, partition, offset, key, value, dan timestamp. Loop while (true) adalah pola standar — satu thread satu consumer.

#Consumer Lifecycle

Consumer melewati beberapa fase: assign partition (join group), membaca record dari offset yang disimpan, melakukan commit, dan akhirnya close() yang melepas partition dan mengirim final commit. Kegagalan menutup consumer dengan benar dapat menahan partition sampai session timeout.

#Consumer Groups

#Group Coordination

Setiap group memiliki group coordinator — salah satu broker — yang melacak anggota group dan partition assignment. Saat anggota join atau leave, koordinator memicu rebalance: menghentikan konsumsi semua anggota, menghitung ulang assignment, dan melanjutkan. Ini transparan tetapi mahal, jadi perlu diminimalkan.

#Partition Assignment Strategies

Strategi assignment menentukan bagaimana partition dibagi antar anggota:

• Range: membagi partition per topic secara berurutan; dapat membuat ketidakseimbangan di banyak topic.
• RoundRobin: menyebar partition semua topic secara bergantian, hasil lebih merata.
• Sticky: mempertahankan assignment sebelumnya sebanyak mungkin saat rebalance.
• CooperativeSticky (3.1+): rebalance inkremental — hanya partition yang berubah yang dipindahkan, tanpa penghentian total.

partition.assignment.strategy=org.apache.kafka.clients.consumer.CooperativeStickyAssignor

#Rebalancing dan Static Membership

Rebalance penuh (eager) menghentikan semua anggota sesaat. CooperativeSticky menghindari hal ini, dan static membership (group.instance.id) membuat consumer tetap dianggap anggota meski restart singkat, mencegah rebalance saat deployment rolling. Kombinasi keduanya membuat group stabil di produksi.

#Offset Management

#Auto-Commit vs Manual Commit

Secara default enable.auto.commit=true dan auto.commit.interval.ms=5000: offset di-commit otomatis setiap 5 detik. Risikonya adalah at-least-once — jika aplikasi crash antara proses dan commit, record akan diproses ulang. Untuk kontrol penuh, matikan auto-commit dan commit manual setelah proses selesai:

enable.auto.commit=false

for (ConsumerRecord<String, String> r : records) {
    process(r);
}
consumer.commitSync();

commitSync() memblokir sampai commit berhasil; commitAsync() mengembalikan hasil via callback dan tidak memblokir loop. Pola terbaik: commitAsync() di loop normal, lalu commitSync() saat close sebagai jaring pengaman terakhir.

#Offset Storage dan Seeking

Offset consumer disimpan di topic internal __consumer_offsets, dikelola oleh koordinator group. Kalian dapat mengabaikan posisi tersimpan dan membaca dari awal (seekToBeginning), akhir (seekToEnd), atau offset spesifik. auto.offset.reset menentukan perilaku saat group belum punya offset:

• earliest: baca dari awal partition.
• latest: baca hanya record baru (default).
• none: error jika offset tidak ditemukan.

#Consumer Configuration Tuning

#Fetch Parameters

fetch.min.bytes dan fetch.max.wait.ms menyeimbangkan throughput dan latensi: broker menahan respons sampai minimal N byte terkumpul atau menunggu maksimal M ms. max.poll.records membatasi jumlah record per poll() — penting untuk menjaga waktu pemrosesan tetap dalam max.poll.interval.ms.

#Timeout dan Heartbeat

• session.timeout.ms (default 45s): broker menganggap consumer mati jika tidak ada heartbeat sejauh ini.
• heartbeat.interval.ms: frekuensi heartbeat, biasanya sepertiga dari session timeout.
• max.poll.interval.ms (default 5 menit): batas waktu total memproses hasil satu poll(); jika terlampaui, consumer dikeluarkan dari group.

#Pola Consumer Lanjutan

Beberapa pola yang sering dipakai:

• Satu consumer per partition untuk ordering dan isolation maksimal.
• Consumer group untuk paralelisme horizontal; tambah anggota sampai jumlah partition.
• Manual partition assignment dengan assign() ketika kalian tidak ingin koordinator menentukan sendiri.
• Pause/resume (pause() dan resume()) untuk backpressure saat downstream melambat.

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-processor --describe

Output menampilkan per partition: current-offset, log-end-offset, lag, dan consumer yang memegang partition. kafka-consumer-groups.sh --describe ini akan menjadi tool rutin untuk memantau kemajuan konsumsi.

#Penutup

Di episode 6 ini kalian sudah menguasai consumer: subscription dan polling, lifecycle, consumer group dengan strategi assignment Range, RoundRobin, Sticky, dan CooperativeSticky, pengelolaan offset dengan auto-commit dan manual commit, plus tuning fetch.min.bytes, max.poll.records, dan parameter timeout.

Inti yang harus dibawa pulang:

• poll() sekaligus mengambil record dan mengirim heartbeat.
• Satu partition hanya dibaca satu consumer dalam satu group; tambah consumer hingga jumlah partition.
• CooperativeSticky dan static membership meminimalkan dampak rebalance.
• Commit offset setelah proses selesai; pakai commitAsync plus commitSync saat close.
• Offset disimpan di topic __consumer_offsets.
• auto.offset.reset=earliest untuk membaca dari awal, latest untuk record baru saja.

Di episode 7 selanjutnya kita akan menyiapkan data yang dikirim: message serialization dan schema management — dari String, JSON, Avro, dan Protobuf, hingga peran Confluent Schema Registry dalam schema evolution dan compatibility types. Kalian akan belajar mendefinisikan skema Avro dan memilih format yang tepat untuk aplikasi kalian!