Implementasi producer dan consumer

Dokumen arsitektur proyek sensor suhu sudah menetapkan peta topic, skema payload, dan kontrak pengiriman, sehingga tahap berikutnya menerjemahkannya menjadi kode klien yang benar-benar berjalan. Topik ini menuliskan dua layanan minimum proyek secara utuh, yaitu layanan perekam sebagai producer kejadian dan layanan perekap sebagai consumer yang menghitung rangkuman, yang keduanya sadar lag (lag-aware) sesuai kontrak keberhasilan proyek. Kedua layanan dikerjakan sebagai dua skrip Python utuh yang dapat dijalankan pada broker lokal, dan setiap keputusan implementasinya dikaitkan kembali ke kontrak pengiriman yang sudah ditetapkan pada dokumen arsitektur.

Layanan perekam dengan producer idempoten#

Layanan perekam menerjemahkan setiap pembacaan sensor menjadi kejadian envelope, yaitu bungkus metadata standar yang sudah disepakati pada skema payload, lalu mengirimkannya dengan jaminan pengiriman tertinggi sebelum rangkuman dihitung. Kontrak pengiriman menetapkan semantik at-least-once pada alur ini, sehingga producer menunggu konfirmasi dari semua in-sync replica (ISR) dan broker menolak duplikat nomor urut dari sesi producer yang sama. Dua konfigurasi klien yang mewujudkan kebijakan itu adalah acks="all" dan producer idempoten melalui enable_idempotence=True. Skrip berikut merupakan layanan perekam lengkap yang siap dijalankan.

# layanan perekam proyek sensor suhu: producer idempoten dengan acks=all
import json
import time
import uuid
from datetime import datetime, timezone

from kafka import KafkaAdminClient, KafkaProducer
from kafka.admin import NewTopic
from kafka.errors import TopicAlreadyExistsError

BROKER = "localhost:9092"
topic = f"m8-perekam-{int(time.time())}"  # nama unik per eksekusi

# topic demo dibuat lebih dahulu, meniru pembuatan m8-suhu-kejadian
admin = KafkaAdminClient(bootstrap_servers=BROKER)
try:
    admin.create_topics([NewTopic(name=topic, num_partitions=3, replication_factor=1)])
except TopicAlreadyExistsError:
    pass
admin.close()

producer = KafkaProducer(
    bootstrap_servers=BROKER,
    acks="all",                # konfirmasi dari semua ISR sebelum dianggap terkirim
    enable_idempotence=True,   # producer idempoten sesuai kontrak pengiriman
)

perangkat = [
    ("sensor-01", 1, (24.5, 24.7)),
    ("sensor-02", 2, (23.8, 24.1)),
    ("sensor-03", 3, (22.9, 23.2)),
    ("sensor-04", 4, (25.3, 25.1)),
]

total = 0
for id_perangkat, lantai, bacaan in perangkat:
    for suhu in bacaan:
        kejadian = {
            "event_id": str(uuid.uuid4()),
            "event_type": "sensor.suhu.terbaca",
            "occurred_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
            "data": {"id_perangkat": id_perangkat, "lantai": lantai, "suhu_c": suhu},
        }
        md = producer.send(topic, key=id_perangkat.encode(),
                           value=json.dumps(kejadian).encode()).get(timeout=10)
        print(f"kirim key={id_perangkat} -> partisi={md.partition} offset={md.offset}")
        total += 1

producer.flush()
producer.close()
print(f"{total} kejadian terkirim ke topic demo")

kirim key=sensor-01 -> partisi=0 offset=0
kirim key=sensor-01 -> partisi=0 offset=1
kirim key=sensor-02 -> partisi=0 offset=2
kirim key=sensor-02 -> partisi=0 offset=3
kirim key=sensor-03 -> partisi=1 offset=0
kirim key=sensor-03 -> partisi=1 offset=1
kirim key=sensor-04 -> partisi=1 offset=2
kirim key=sensor-04 -> partisi=1 offset=3
8 kejadian terkirim ke topic demo

Keluaran memperlihatkan penempatan setiap kejadian secara eksplisit, yaitu nomor partisi dan offset hasil metadatanya. Offset mulai dari 0 pada setiap partisi karena demo membuat topic baru dengan KafkaAdminClient agar skrip berdiri sendiri, sedangkan pada proyek nyata topic m8-suhu-kejadian sudah dibuat lebih dahulu dan perekam cukup mengirim kejadian. Penomoran yang mulai dari 0 juga menegaskan keuntungan topic baru untuk demo, yaitu setiap angka pada keluaran dapat diperiksa ulang pembaca dengan topic miliknya sendiri.

Nilai yang dicetak dipilih dengan sengaja agar deterministik. Field event_id dan occurred_at yang berada di dalam envelope memang berubah pada setiap eksekusi, sehingga keduanya tidak dicetak, sedangkan id_perangkat, partisi, dan offset sepenuhnya ditentukan oleh key serta topic baru. Key sensor-01 dan sensor-02 masuk ke partisi 0, sedangkan sensor-03 dan sensor-04 masuk ke partisi 1, sesuai perhitungan murmur2 yang digunakan klien untuk memetakan key ke partisi.

Layanan perekap dengan dedupe dan commit eksplisit#

Layanan perekap membaca aliran kejadian, menghitung rangkumannya, menyimpan hasilnya, dan baru setelah semuanya selesai menandai kemajuan konsumsi. Urutan penyimpanan dahulu dan commit belakangan itulah yang menjaga kontrak at-least-once tidak terbalik menjadi kehilangan kejadian. Untuk itu auto commit dimatikan melalui enable_auto_commit=False, dan commit() dipanggil oleh kode setelah seluruh batch selesai diproses. Penyaring dedupe, yaitu pengecekan event_id sebelum kejadian dihitung, menutup sisa celah kontrak karena kejadian dapat tiba dua kali ketika pemrosesan berhasil tetapi commitnya gagal. Skrip berikut memuat seluruh layanan tersebut. Untuk keperluan demo, topic beserta kejadian awal disiapkan sendiri oleh skrip, termasuk satu kejadian yang dikirim ulang dengan event_id yang sama, sehingga eksekusinya berdiri sendiri tanpa menjalankan perekam terlebih dahulu.

# layanan perekap proyek sensor suhu: dedupe event_id dan commit eksplisit
import json
import time
import uuid
from datetime import datetime, timezone

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

BROKER = "localhost:9092"
topic = f"m8-perekap-{int(time.time())}"      # nama unik per eksekusi
grup = f"grup-m8-perekap-{int(time.time())}"  # group baru agar offset bersih

# topic demo beserta kejadian awalnya disiapkan agar skrip berdiri sendiri
admin = KafkaAdminClient(bootstrap_servers=BROKER)
try:
    admin.create_topics([NewTopic(name=topic, num_partitions=3, replication_factor=1)])
except TopicAlreadyExistsError:
    pass
admin.close()


def buat_kejadian(id_perangkat, lantai, suhu, event_id=None):
    return {
        "event_id": event_id or str(uuid.uuid4()),
        "event_type": "sensor.suhu.terbaca",
        "occurred_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "data": {"id_perangkat": id_perangkat, "lantai": lantai, "suhu_c": suhu},
    }


perangkat = [
    ("sensor-01", 1, (24.5, 24.7, 24.6)),
    ("sensor-02", 2, (23.8, 24.1, 23.9)),
    ("sensor-03", 3, (22.9, 23.2, 23.0)),
    ("sensor-04", 4, (25.3, 25.1, 25.4)),
]

semua = []
for id_perangkat, lantai, bacaan in perangkat:
    for suhu in bacaan:
        semua.append(buat_kejadian(id_perangkat, lantai, suhu))
# satu kejadian dikirim ulang dengan event_id yang sama, meniru redelivery
semua.append(buat_kejadian("sensor-01", 1, 24.5, event_id=semua[0]["event_id"]))

producer = KafkaProducer(bootstrap_servers=BROKER)
for kejadian in semua:
    producer.send(topic, key=kejadian["data"]["id_perangkat"].encode(),
                  value=json.dumps(kejadian).encode())
producer.flush()
producer.close()

consumer = KafkaConsumer(bootstrap_servers=BROKER, group_id=grup,
                         auto_offset_reset="earliest", enable_auto_commit=False)
partisi = sorted(consumer.partitions_for_topic(topic))
tps = [TopicPartition(topic, p) for p in partisi]
# partisi ditetapkan manual agar demo deterministik, perekap nyata menggunakan
# mode berlangganan sehingga partisi dibagi otomatis antar anggota group
consumer.assign(tps)
diproses = set()   # event_id yang sudah lolos, disimpan bersama rangkuman di nyata
hasil = []
dilewati = []


def proses(rec):
    kejadian = json.loads(rec.value.decode())
    data = kejadian["data"]
    eid = kejadian["event_id"]
    if eid in diproses:
        dilewati.append((rec.partition, rec.offset, data["id_perangkat"]))
        return
    diproses.add(eid)
    hasil.append((rec.partition, rec.offset, data["id_perangkat"], data["suhu_c"]))


def ukur_lag(label):
    akhir = consumer.end_offsets(tps)
    print(label)
    for tp in tps:
        committed = consumer.committed(tp) or 0
        end = akhir[tp]
        print(f"  partisi={tp.partition} end_offset={end} committed={committed} lag={end - committed}")


ukur_lag("lag sebelum perekap memproses:")

# antrean dihabiskan, commit dilakukan setelah setiap batch selesai diproses
while True:
    batch = consumer.poll(timeout_ms=1000, max_records=100)
    if not batch:
        break
    for records in batch.values():
        for rec in records:
            proses(rec)
    consumer.commit()

print("kejadian per partisi:")
for part, offset, perangkat_id, suhu in sorted(hasil):
    print(f"  partisi={part} offset={offset} key={perangkat_id} suhu={suhu} -> disimpan")
for part, offset, perangkat_id in sorted(dilewati):
    print(f"  partisi={part} offset={offset} key={perangkat_id} duplikat -> dilewati")
ukur_lag("lag setelah antrean dihabiskan:")
consumer.close()

lag sebelum perekap memproses:
  partisi=0 end_offset=7 committed=0 lag=7
  partisi=1 end_offset=6 committed=0 lag=6
  partisi=2 end_offset=0 committed=0 lag=0
kejadian per partisi:
  partisi=0 offset=0 key=sensor-01 suhu=24.5 -> disimpan
  partisi=0 offset=1 key=sensor-01 suhu=24.7 -> disimpan
  partisi=0 offset=2 key=sensor-01 suhu=24.6 -> disimpan
  partisi=0 offset=3 key=sensor-02 suhu=23.8 -> disimpan
  partisi=0 offset=4 key=sensor-02 suhu=24.1 -> disimpan
  partisi=0 offset=5 key=sensor-02 suhu=23.9 -> disimpan
  partisi=1 offset=0 key=sensor-03 suhu=22.9 -> disimpan
  partisi=1 offset=1 key=sensor-03 suhu=23.2 -> disimpan
  partisi=1 offset=2 key=sensor-03 suhu=23.0 -> disimpan
  partisi=1 offset=3 key=sensor-04 suhu=25.3 -> disimpan
  partisi=1 offset=4 key=sensor-04 suhu=25.1 -> disimpan
  partisi=1 offset=5 key=sensor-04 suhu=25.4 -> disimpan
  partisi=0 offset=6 key=sensor-01 duplikat -> dilewati
lag setelah antrean dihabiskan:
  partisi=0 end_offset=7 committed=7 lag=0
  partisi=1 end_offset=6 committed=6 lag=0
  partisi=2 end_offset=0 committed=0 lag=0

Keluaran pertama menunjukkan backlog penuh sebelum perekap memproses apa pun, yaitu committed 0 pada semua partisi sedangkan ujung lognya berisi 7 dan 6 kejadian. Pemrosesan kemudian menghabiskan antrean, dan daftar prosesnya dikelompokkan per partisi agar urutan antar partisi tidak menyembunyikan baris yang sebenarnya berurutan dalam log. Kejadian pada partisi 0 offset 6 adalah kejadian ulang yang event_id-nya sudah pernah diproses, sehingga barisnya bertanda duplikat dan tidak masuk ke penyimpanan. Lag akhir bernilai 0 pada semua partisi menandakan commit mengikuti ujung log. Penetapan partisi manual pada demo digunakan agar keluarannya deterministik, sedangkan perekap proyek berjalan dengan mode berlangganan sehingga partisinya dibagi otomatis antar anggota group.

Skrip demo menyimpan event_id yang sudah diproses dalam himpunan di memori, dan himpunan itu cukup untuk satu eksekusi. Pada proyek nyata, himpunan yang sama harus bertahan antar restart, karena perekap yang dihidupkan ulang tetap menerima kejadian yang mungkin sudah dihitung sebelumnya. Ukurannya juga tumbuh mengikuti aliran, sehingga jangkauan waktunya perlu dibatasi secara eksplisit, misalnya per hari per perangkat.

Pemantauan lag di dalam layanan perekap#

Klien Python tidak menyediakan metode lag tersendiri, tetapi kedua angka penyusunnya tersedia, dan layanan yang sadar lag menyusunnya sendiri. Fungsi ukur_lag pada perekap membaca ujung log dengan end_offsets() dan posisi commit dengan committed() untuk setiap partisi yang ditugaskan, lalu mencetak selisihnya sebagai lag. Pengukuran dipanggil dua kali pada demo, yaitu sebelum antrean diproses dan setelah antrean habis, sehingga keduanya memperlihatkan keadaan berbeda dari kode yang sama.

Kontrak keberhasilan proyek menyatakan backlog perekap tidak melewati sepuluh menit, dan angka lag baru bermakna terhadap kontrak itu setelah dikonversi menjadi durasi. Laju kejadian yang dihitung pada tahap pemilihan proyek adalah 0.33 kejadian per detik, sehingga lag berisi 20 kejadian berarti backlog sekitar satu menit, sedangkan ambang sepuluh menit setara sekitar 200 kejadian pada laju yang sama. Konversi sederhana seperti ini cukup dijadikan logika peringatan pada perekap, yaitu menghitung durasi setiap kali lag diukur dan memberi tanda apabila lag melewati ambangnya. Tren lag tetap lebih menentukan daripada satu sampel, karena lonjakan sesaat adalah hal yang normal dan kenaikan terus-menerus barulah keadaan yang menuntut penanganan.

Keputusan implementasi menurut kontrak#

Setiap konfigurasi pada kedua layanan dapat dilacak balik ke satu ketentuan kontrak, sehingga tabel berikut berfungsi sebagai catatan keputusan implementasi yang memperlihatkan rantai itu. Baris pertama menahan duplikat di sisi log, baris kedua menahannya di sisi perekap, baris ketiga menjaga kejadian tidak hilang, dan baris keempat membuat seluruhnya terukur. Alasan pada kolom terakhir mengutip ketentuan dokumen arsitektur, sehingga perubahan kontrak selalu terlihat sampai ke konfigurasinya. Keempat barisnya saling menopang, dan menggugurkan satu baris membuat baris lainnya kehilangan alasan keberadaannya.

Rantai tersebut juga menjadi daftar periksa saat perekap direvisi. Perubahan kecil seperti menaikkan max.poll.records tidak menyentuh keempat keputusan, sedangkan perubahan pada urutan penyimpanan dan commit menyentuh dua baris sekaligus, sehingga perubahan semacam itu ditinjau seperti revisi kontrak. Cara kerja ini menjaga implementasi tetap mengikuti dokumen arsitektur ketika proyek tumbuh.

Kesalahan umum#

1. Menyalakan auto commit agar kode consumer lebih pendek. Posisi commit ikut naik bersama pemanggilan poll, bahkan sebelum batch selesai disimpan, sehingga perekap yang mati di tengah batch kehilangan kejadian. Matikan auto commit dan panggil commit() setelah penyimpanan, sebagaimana perekap pada topik ini.
2. Menganggap producer idempoten sudah menyelesaikan dedupe. Jaminan idempotence hanya berlaku pada satu sesi producer, sedangkan kejadian yang dikirim ulang oleh sesi berbeda tetap tercatat dua kali di log. Pengecekan event_id pada perekap tetap diperlukan karena jaminannya berhenti di log Kafka.
3. Menyimpan himpunan event_id di memori pada layanan yang dihidupkan ulang. Himpunan itu kosong kembali setiap restart, sehingga rangkuman dihitung dua kali untuk kejadian yang sama. Simpan event_id bersama hasil rangkuman agar pemeriksaannya bertahan antar restart.
4. Melaporkan lag dalam jumlah kejadian mentah tanpa mengaitkannya dengan laju topic. Angka 200 kejadian berarti backlog satu menit pada laju 0.33 kejadian per detik, tetapi jauh lebih panjang pada laju yang lebih kecil. Ubah lag menjadi durasi sebelum dibandingkan dengan ambang kontrak.

Ringkasan#

• Layanan perekam mengirim kejadian envelope berunci id_perangkat dengan acks="all" dan producer idempoten, dan keluarannya menampilkan partisi serta offset setiap kejadian.
• Layanan perekap memproses dengan dedupe event_id, menyimpan hasilnya, lalu commit offset setelah batch selesai, sehingga kontrak at-least-once terjaga sampai penyimpanan.
• Pengukuran lag disusun dari end_offsets() dan committed(), dan angkanya dikonversi menjadi durasi untuk dibandingkan dengan kontrak backlog sepuluh menit.
• Tabel keputusan memetakan setiap konfigurasi klien ke ketentuan kontraknya, sehingga revisi implementasi selalu melewati peninjauan yang sama.

Referensi#

• Producer configs — Apache Software Foundation — konfigurasi acks dan enable.idempotence yang menjadi dasar layanan perekam.
• Consumer configs — Apache Software Foundation — konfigurasi enable.auto.commit dan auto.offset.reset yang digunakan layanan perekap.
• Basic Kafka operations — Apache Software Foundation — perintah pemantauan offset dan lag group yang melengkapi pengukuran dari kode.
• kafka-python KafkaProducer — kafka-python project — metode send beserta metadata partisi dan offset yang dicetak demo.
• kafka-python KafkaConsumer — kafka-python project — metode end_offsets dan committed yang menyusun pengukuran lag.