Consumer lag

Consumer lag adalah ukuran ketertinggalan sebuah consumer group terhadap ujung log, dihitung dari selisih antara log-end-offset dan offset committed. Topik ini membahas cara lag itu terbentuk, pengukuran lewat CLI dan lewat klien Python, kebiasaan menetapkan ambang peringatan, serta pola pikir backpressure ketika lag terus bertambah. Demo dijalankan pada broker lokal dengan topic dan group milik modul ini, sehingga angka pada keluaran dapat direproduksi.

Selisih dua offset yang disebut lag#

Setiap partisi menyimpan pesan dengan offset berurutan, dan dua posisi pada log itu menentukan lag. Log-end-offset adalah offset pesan berikutnya yang akan ditulis broker, sedangkan offset committed adalah posisi terakhir yang di-commit oleh consumer group, yaitu offset pesan berikutnya yang akan dibaca konsumennya. Selisih keduanya adalah jumlah pesan yang sudah berada di log tetapi belum diproses group.

Lag 0 berarti konsumen membaca sampai ujung log, sedangkan lag yang bertahan di atas 0 berarti ada pesan yang menunggu pemrosesan. Lag diukur per partisi karena kedua posisinya berada pada satu partisi, dan lag sebuah group adalah jumlah lag seluruh partisi yang ditugaskan ke group tersebut. Diagram berikut memperlihatkan posisi keduanya pada satu partisi berisi lima pesan, dalam keadaan konsumen baru memproses pesan pertama.

Karena lag adalah selisih dua offset, angkanya berubah terus pada sistem yang hidup. Producer menaikkan log-end-offset setiap pesan baru masuk, sedangkan commit konsumen menaikkan sisi committed dengan jeda yang tergantung pada strategi commitnya. Pengukuran sekali sesaat hanya menangkap keadaan satu momen, sehingga kebiasaan membaca lag yang sehat mengandalkan tren, bukan satu sampel.

Mengukur lag lewat CLI#

Perintah bin/kafka-consumer-groups.sh --describe --group <nama> adalah alat bawaan distribusi Kafka untuk melihat lag sebuah group. Keluarannya berupa tabel per partisi dengan kolom CURRENT-OFFSET untuk posisi committed, LOG-END-OFFSET untuk ujung log, dan LAG untuk selisihnya. Demo berikut menyiapkan topic tetap m4-lag-demo yang dibuat baru pada setiap eksekusi, dan group m4-lag-grup dihapus di awal agar offset committed mulai dari kosong sehingga bentuk tabelnya selalu sama.

(
  cd /home/kal/my-project/kafka-runtime/kafka
  bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group m4-lag-grup >/dev/null 2>&1
  bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --if-exists --topic m4-lag-demo >/dev/null 2>&1
  sleep 2
  bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic m4-lag-demo --partitions 1
)

Created topic m4-lag-demo.

Topic pada blok di atas masih kosong. Blok berikut mengisinya dengan lima pesan, lalu konsumen group m4-lag-grup sengaja hanya memproses satu pesan lewat --max-messages 1, dan lagnya diukur dengan describe pada akhir blok yang sama.

(
  cd /home/kal/my-project/kafka-runtime/kafka
  printf "bacaan-lag-0\nbacaan-lag-1\nbacaan-lag-2\nbacaan-lag-3\nbacaan-lag-4\n" | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic m4-lag-demo 2>/dev/null
  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic m4-lag-demo --group m4-lag-grup --from-beginning --max-messages 1 2>/dev/null
  bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group m4-lag-grup 2>/dev/null
)

bacaan-lag-0

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
m4-lag-grup     m4-lag-demo     0          1               5               4               -               -               -

Baris pertama keluaran adalah pesan yang dicetak konsumen konsol, yaitu bacaan-lag-0 sebagai satu-satunya pesan yang diproses. Konsumen konsol meng-commit offsetnya saat keluar secara normal, sehingga CURRENT-OFFSET berisi 1. Tabel describe selalu diawali satu baris kosong sebagai perilaku bawaan alat, dan LOG-END-OFFSET berisi 5 karena lima pesan sudah tercatat di log. Kolom LAG pun berisi 4, persis selisih keduanya, sedangkan tiga kolom terakhir berisi - karena anggota group yang memproses sudah keluar sehingga tidak ada konsumen aktif yang dilaporkan.

Menghitung lag dengan klien Python#

Klien Python tidak menyediakan satu metode lag, tetapi menyediakan kedua angka penyusunnya. Metode end_offsets() mengembalikan log-end-offset untuk sekumpulan partisi, sedangkan committed() mengembalikan offset committed sebuah partisi untuk group konsumen tersebut, dan selisihnya dihitung di sisi aplikasi. Skrip berikut membuat topic unik berisi lima pesan, memproses satu pesan lalu commit, mengukur lag, menghabiskan sisa antrean, dan mengukur lagi.

from kafka import KafkaConsumer, KafkaProducer, KafkaAdminClient, TopicPartition
from kafka.admin import NewTopic
from kafka.errors import TopicAlreadyExistsError
import time

topic = f"m4-lag-py-{int(time.time())}"  # nama unik per eksekusi

admin = KafkaAdminClient(bootstrap_servers="localhost:9092")
try:
    admin.create_topics([NewTopic(name=topic, num_partitions=1, replication_factor=1)])
except TopicAlreadyExistsError:
    pass
admin.close()

# lima pesan dimasukkan sebagai beban antrean
producer = KafkaProducer(bootstrap_servers="localhost:9092")
for i in range(5):
    producer.send(topic, value=f"bacaan-{i}".encode())
producer.flush()
producer.close()


def ukur_lag(konsumen, partisi):
    akhir = konsumen.end_offsets([partisi])[partisi]
    tercommit = konsumen.committed(partisi)
    print(f"log-end-offset={akhir}, committed={tercommit}, lag={akhir - tercommit}")


consumer = KafkaConsumer(topic, bootstrap_servers="localhost:9092",
                         group_id=f"m4-lag-grup-py-{int(time.time())}",
                         auto_offset_reset="earliest", enable_auto_commit=False)
tp = TopicPartition(topic, 0)

# konsumen lambat: hanya satu pesan yang diproses lalu di-commit
consumer.poll(timeout_ms=5000, max_records=1)
consumer.commit()
print("setelah satu pesan diproses:")
ukur_lag(consumer, tp)

# konsumen mengejar ketertinggalan sampai tidak ada pesan baru
while consumer.poll(timeout_ms=1000, max_records=500):
    pass
consumer.commit()
print("setelah antrean dihabiskan:")
ukur_lag(consumer, tp)
consumer.close()

setelah satu pesan diproses:
log-end-offset=5, committed=1, lag=4
setelah antrean dihabiskan:
log-end-offset=5, committed=5, lag=0

Angka pertama menunjukkan keadaan konsumen lambat, yaitu satu pesan diproses sementara lima pesan sudah menunggu di log. Iterasi berikutnya menghabiskan antrean, commit dijalankan lagi, dan lag turun menjadi 0 karena posisi committed mengikuti ujung log. Skrip menggunakan group baru pada setiap eksekusi dengan enable_auto_commit=False, sehingga kedua pengukuran sepenuhnya dikendalikan oleh kode demo dan tidak terpengaruh offset lama.

Ambang peringatan dan backpressure#

Lag yang berguna sebagai alarm adalah trennya, bukan nilai sesaat. Lonjakan yang kembali turun dalam hitungan detik adalah bentuk normal, misalnya saat producer mengirim gumpalan besar atau saat konsumen baru saja dijalankan ulang, karena konsumen yang sehat mengejar dalam waktu singkat. Lag yang naik terus dari satu pemeriksaan ke pemeriksaan berikutnya berarti laju konsumsi lebih kecil daripada laju produksi, dan keadaan itu tidak pulih dengan sendirinya.

Ambang peringatan yang tepat berhubungan dengan waktu, bukan dengan jumlah mentah. Lag 5000 pesan berarti backlog lima detik pada laju masuk 1000 pesan per detik, tetapi berarti backlog lima puluh detik pada laju 100 pesan per detik. Karena itu, ambang lazim dinyatakan sebagai rasio lag terhadap laju masuk, misalnya memberi peringatan ketika backlog melewati sepuluh detik, dan angka ambangnya diturunkan dari laju setiap topic.

Respons atas lag yang tidak turun mengikuti pola pikir backpressure, yaitu penyesuaian laju antara producer dan konsumen sampai keduanya seimbang. Kapasitas sisi konsumen dapat ditingkatkan dengan menambah jumlah konsumen sampai sama dengan jumlah partisi, memperbesar batch pemrosesan, atau menaikkan max.poll.records bila hambatannya ada pada jumlah pesan per pemanggilan poll. Sisi masukan dapat ditahan dengan pembatasan laju pada producer sebelum pesan menumpuk lebih jauh. Pilihan tindakan mengikuti penyebabnya, karena menambah konsumen tidak menolong bila hambatannya ada pada kode pemrosesan, dan menahan masukan hanya menggeser masalah bila antreannya benar-benar kewalahan.

Metrik lag pada klien#

Klien Java mengekspos metrik konsumsi lewat JMX, dan sistem pemantauan mengumpulkannya tanpa harus menjalankan CLI. Tabel berikut memuat metrik yang paling dekat dengan pengukuran lag beserta sumber MBean-nya.

Metrik per partisi juga tersedia pada MBean yang sama dengan tambahan dimensi topic dan partisi, sehingga partisi yang paling tertinggal dapat ditemukan langsung. Karena metrik ini diterbitkan oleh klien dan bukan oleh broker, pengumpulannya dilakukan dari host tempat aplikasi konsumen berjalan. Pada aplikasi yang tidak menggunakan klien Java, pengukuran CLI dan klien Python di atas menjadi sumber lag, dan keduanya dijadwalkan berkala, misalnya setiap sepuluh detik, untuk membentuk tren.

Kesalahan umum#

1. Menetapkan ambang dalam jumlah pesan mentah tanpa melihat laju topic. Angka yang sama berarti durasi tunggu berbeda antar topic dengan laju berbeda, sehingga alarmnya menyesatkan. Nyatakan ambang sebagai durasi backlog yang diturunkan dari laju masuk.
2. Menambah jumlah konsumen melebihi jumlah partisi. Dalam satu group, satu partisi hanya ditugaskan ke satu konsumen, sehingga konsumen tambahan menganggur dan lag tidak turun. Tambahkan partisi terlebih dahulu bila paralelisme yang lebih besar benar-benar diperlukan.
3. Menyimpulkan kondisi dari satu sampel lag. Lonjakan sesaat adalah bentuk normal pada sistem yang hidup, sedangkan keadaan berbahaya tampak dari tren yang naik terus. Simpan sampel berkala dan tinjau kecenderungannya sebelum menindak.
4. Mencoba mengukur lag konsumen yang tidak tergabung dalam group. Perintah describe hanya tersedia untuk group, sedangkan konsumen tanpa group tidak meng-commit sehingga tidak ada posisi yang dapat dibandingkan dengan ujung log. Beri group pada konsumen yang lag-nya dipantau.

Ringkasan#

• Lag adalah selisih log-end-offset dan offset committed group, diukur per partisi dan dijumlahkan per group.
• kafka-consumer-groups --describe menampilkan kedua posisi beserta lag per partisi, sedangkan klien Python menyusunnya dari end_offsets() dan committed().
• Ambang peringatan yang berguna dinyatakan sebagai durasi backlog dan ditinjau dari trennya, bukan dari satu sampel.
• Backpressure menyeimbangkan laju dengan menambah kapasitas konsumen sampai batas partisi atau menahan laju masukan.

Referensi#

• Basic Kafka operations — Apache Software Foundation — perintah kafka-consumer-groups untuk melihat offset dan lag group.
• Monitoring — Apache Software Foundation — metrik consumer seperti records-lag-max dan records-lead-min beserta MBean-nya.
• Consumer configs — Apache Software Foundation — konfigurasi max.poll.records yang memengaruhi jumlah pesan per pemanggilan poll.
• kafka-python KafkaConsumer — kafka-python project — metode end_offsets dan committed yang digunakan pada demo Python.