Offset dan commit

Topik ini membahas cara consumer menandai posisi bacanya di dalam partisi, mulai dari offset yang di-commit, dua strategi commit yang disediakan klien, perintah seeking untuk berpindah posisi, sampai reset offset pada consumer group lewat CLI. Demo Python dan demo CLI dijalankan pada broker nyata dengan keluaran aslinya ditempel pada pembahasannya. Setelah membacanya, posisi baca sebuah group dapat diawasi, digeser, dan diputar ulang dengan sengaja.

Posisi baca yang tersimpan#

Consumer menyimpan posisi bacanya di broker sebagai committed offset, yaitu nomor offset pesan berikutnya yang akan dibaca oleh satu group pada satu partisi. Nilai itu tersimpan di topic internal __consumer_offsets, sehingga group yang mati dan bangkit kembali melanjutkan pekerjaan dari posisi terakhirnya tanpa mengulang seluruh log. Nilainya selalu maju satu langkah dari pesan terakhir yang selesai diproses, karena commit pada offset N menyatakan pesan 0 sampai N-1 sudah diproses.

Dua nilai berbeda perlu dipisahkan sejak awal, yaitu posisi baca saat ini pada klien dan committed offset di broker. Posisi baca bergerak setiap kali poll() mengembalikan pesan, sedangkan committed offset hanya berubah ketika commit dilakukan. Consumer yang bergabung kembali membaca mulai dari committed offset, bukan dari posisi klien lama yang sudah hilang bersama prosesnya. Selisih keduanya muncul sebagai lag, yaitu jumlah pesan yang sudah masuk log tetapi belum dicapai oleh committed offset.

Penempatan commit menentukan semantik pengiriman pada saat kegagalan. Commit sebelum pemrosesan selesai menimbulkan risiko pesan terlewat apabila proses mati seketika setelahnya, sedangkan commit setelah pemrosesan selesai menimbulkan risiko pesan diproses dua kali apabila proses mati sebelum sempat commit. Umumnya sistem memilih risiko duplikasi karena pemrosesan ulang lebih mudah ditangani daripada data yang hilang.

Auto-commit dan commit eksplisit#

Klien Kafka menyediakan dua strategi commit, yaitu auto-commit yang berjalan di latar belakang dan commit eksplisit yang dikendalikan kode pemroses. Auto-commit diaktifkan lewat konfigurasi enable.auto.commit dengan nilai default true, lalu klien meng-commit posisi baca setiap auto.commit.interval.ms berlalu, yaitu 5000 milidetik pada nilai defaultnya. Commit itu dilakukan saat perulangan poll() berjalan, sehingga aplikasi yang mati di antara dua interval mengulang sebagian pesan yang sudah diambilnya.

Commit eksplisit memindahkan keputusan posisi ke dalam kode. Metode commitSync() menunggu konfirmasi broker sebelum kembali, sehingga aman tetapi menahan perulangan, sedangkan commitAsync() mengirim commit tanpa menunggu dan menerima hasilnya lewat callback. Kombinasi yang lazim menggunakan commitAsync() pada perulangan utama demi kecepatan dan commitSync() pada saat consumer ditutup agar posisi terakhir pasti tersimpan.

Demo commit eksplisit dan pemutaran ulang#

Demo berikut menggunakan topic baru dengan satu partisi dan enam pesan, lalu consumer yang auto-commit-nya dimatikan melalui enable_auto_commit=False. Consumer membaca enam pesan, meng-commit offset dengan commit(), mencetak posisi yang tersimpan, kemudian menggeser posisi bacanya ke awal log dengan seek_to_beginning() dan membaca keenam pesan itu sekali lagi. Topic dibuat terlebih dahulu lewat KafkaAdminClient, dan nama topicnya diberi akhiran waktu agar setiap eksekusi memperoleh log yang kosong.

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

topic = f"m1-offset-{int(time.time())}"  # nama unik per eksekusi
group = f"m1-grup-offset-{int(time.time())}"  # group baru 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()

producer = KafkaProducer(bootstrap_servers="localhost:9092")
for i in range(6):
    producer.send(topic, value=f"kejadian-{i}".encode())
producer.flush()
producer.close()

tp = TopicPartition(topic, 0)
consumer = KafkaConsumer(
    bootstrap_servers="localhost:9092",
    group_id=group,
    auto_offset_reset="earliest",
    enable_auto_commit=False,
)
consumer.subscribe([topic])


def baca(jumlah):
    """Kumpulkan tepat jumlah pesan dari partisi tunggal."""
    hasil = []
    while len(hasil) < jumlah:
        batch = consumer.poll(timeout_ms=10000, max_records=jumlah - len(hasil))
        if tp in batch:
            hasil.extend(batch[tp])
    return hasil


print("konsumsi pertama:")
batch = baca(6)
for msg in batch:
    print(f"  offset={msg.offset} nilai={msg.value.decode()}")

consumer.commit(offsets={tp: OffsetAndMetadata(batch[-1].offset + 1, "", -1)})
print(f"offset ter-commit: {consumer.committed(tp)}")

consumer.seek_to_beginning(tp)
print("replay setelah seek ke awal:")
for msg in baca(6):
    print(f"  offset={msg.offset} nilai={msg.value.decode()}")
consumer.close()

konsumsi pertama:
  offset=0 nilai=kejadian-0
  offset=1 nilai=kejadian-1
  offset=2 nilai=kejadian-2
  offset=3 nilai=kejadian-3
  offset=4 nilai=kejadian-4
  offset=5 nilai=kejadian-5
offset ter-commit: 6
replay setelah seek ke awal:
  offset=0 nilai=kejadian-0
  offset=1 nilai=kejadian-1
  offset=2 nilai=kejadian-2
  offset=3 nilai=kejadian-3
  offset=4 nilai=kejadian-4
  offset=5 nilai=kejadian-5

Baris offset ter-commit: 6 menegaskan makna committed offset sebagai penanda berikutnya, yaitu pesan 0 sampai 5 sudah selesai diproses. Pemutaran ulang dengan seek menghasilkan enam pesan yang persis sama karena posisi baca bergerak kembali ke awal log, sedangkan posisi yang tersimpan di broker tetap 6 sampai commit baru dilakukan. Konsumsi oleh group yang sama dapat diputar ulang kapan pun dengan cara ini selama retensi topic masih menyimpan pesannya.

Seeking dan reset offset#

Perpindahan posisi dari sisi klien disebut seeking, yaitu menggeser posisi baca consumer ke offset tertentu tanpa menyentuh posisi yang tersimpan. Metode seek_to_beginning() pada demo sebelumnya adalah salah satu bentuknya, dan metode seek(tp, offset) mengarahkan posisi ke nomor offset mana pun yang dihitung aplikasi, misalnya untuk mengulang satu batch yang gagal diproses. Seeking hanya berlaku pada consumer yang sedang berjalan, sedangkan posisi group yang sudah ada di broker diatur dari sisi operasi.

Posisi group diatur ulang dengan kafka-consumer-groups.sh --reset-offsets. Perintah ini hanya berlaku untuk group tanpa anggota aktif, karena posisi anggota yang sedang berjalan akan ditimpa lagi oleh commit dari perulangan poll()-nya. Target posisinya dinyatakan sebagai argumen, misalnya --to-earliest untuk mundur ke awal log, --to-latest untuk maju ke ujung log, atau --shift-by N untuk bergerak relatif sejauh N pesan. Demo berikut menggunakan topic tetap m1-offset-demo yang selalu dihapus dan dibuat ulang, sehingga jumlah pesan di dalam lognya selalu sama antar eksekusi.

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

Created topic m1-offset-demo.

Blok berikut mengisi log dengan tiga pesan, membacanya dengan consumer CLI sebagai anggota group m1-grup-reset, lalu mendeskripsikan posisi groupnya. Consumer berhenti sendiri karena opsi --max-messages 3, dan tabel yang dihasilkan menunjukkan keadaan group yang sudah selesai konsumsi, yaitu CURRENT-OFFSET 3 dengan LAG 0.

(
  cd /home/kal/my-project/kafka-runtime/kafka
  printf "pagi\nsiang\nmalam\n" | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic m1-offset-demo 2>/dev/null
  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic m1-offset-demo --group m1-grup-reset --from-beginning --max-messages 3 2>/dev/null
  bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group m1-grup-reset 2>/dev/null
)

pagi
siang
malam

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
m1-grup-reset   m1-offset-demo  0          3               3               0               -               -               -

Perintah reset memundurkan posisi group ke awal log dengan --to-earliest --execute, lalu perintah deskripsi yang sama memperlihatkan akibatnya. Nilai CURRENT-OFFSET kembali ke 0, sedangkan LOG-END-OFFSET tetap 3, sehingga LAG menjadi 3 dan group akan membaca ulang seluruh pesan pada konsumsi berikutnya. Inilah bentuk replay yang dikendalikan dari sisi operasi tanpa menyentuh kode aplikasi.

(
  cd /home/kal/my-project/kafka-runtime/kafka
  bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --reset-offsets --group m1-grup-reset --topic m1-offset-demo --to-earliest --execute 2>/dev/null
  bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group m1-grup-reset 2>/dev/null
)

GROUP           TOPIC           PARTITION  NEW-OFFSET
m1-grup-reset   m1-offset-demo  0          0

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
m1-grup-reset   m1-offset-demo  0          0               3               3               -               -               -

Kesalahan umum#

1. Auto-commit dibiarkan aktif pada pemrosesan yang menulis ke sistem lain, lalu proses mati di tengah pemrosesan sehingga pesan terlewat tanpa jejak. Perbaikannya dengan enable.auto.commit=false, commit setelah setiap batch selesai diproses, dan commitSync() sebelum consumer ditutup.
2. Consumer baru bergabung tanpa posisi tersimpan dan hanya menerima pesan baru karena auto.offset.reset berada pada nilai default latest, sehingga aliran awal tampak tidak pernah terbaca. Atur auto.offset.reset=earliest apabila log awal perlu dibaca, atau reset posisi group dengan --reset-offsets --to-earliest.
3. Perintah --reset-offsets dijalankan pada group yang anggotanya masih hidup sehingga posisi baru itu tertimpa lagi oleh commit anggota yang sedang berjalan. Hentikan seluruh anggota group terlebih dahulu, jalankan perintah reset, kemudian nyalakan anggotanya kembali.

Ringkasan#

• Committed offset tersimpan di topic __consumer_offsets per pasangan group dan partisi, dan nilainya menyatakan pesan berikutnya yang akan dibaca.
• Auto-commit meng-commit posisi setiap lima detik, sedangkan commit eksplisit dengan commitSync() atau commitAsync() mengikuti ritme pemrosesan.
• Seeking menggeser posisi baca klien untuk pemutaran ulang tanpa menyentuh posisi yang tersimpan di broker.
• --reset-offsets --to-earliest mengatur ulang posisi group dari sisi operasi dan hanya berlaku pada group tanpa anggota aktif.

Referensi#

• Consumer configs — Apache Software Foundation — konfigurasi enable.auto.commit, auto.commit.interval.ms, dan auto.offset.reset yang digunakan pada topik ini.
• Apache Kafka documentation — Apache Software Foundation — dokumentasi utama yang menjelaskan cara offset di-commit dalam consumer group.
• KafkaConsumer — kafka-python project — dokumentasi metode commit(), committed(), dan seek_to_beginning() yang digunakan demo Python.