Topic sudah menunggu; sekarang kita mengisi sisi penulisnya. Producer Kafka bertanggung jawab atas tiga hal sebelum sebuah pesan menempel di log: melakukan serialisasi nilai menjadi bytes, memilih partition tujuan, lalu menyerahkan pesan ke broker dan melaporkan hasilnya secara asinkron. Bagian ini membangun producer proyek dengan bentuk API yang mengikuti kafka-python — send() mengembalikan future, flush() memaksa pengiriman — lalu membungkusnya dalam SimulatorCheckout, tiruan halaman checkout yang menerbitkan event pesanan TokoKita. Mekanika acks dan record batch sudah dibedah pada chapter Producers dan Consumers; kode di sini adalah penerapannya.

Satu Protokol, Dua Backend #

Producer dan consumer proyek ini ditulis menghadap protokol kecil — send(topic, key, value), flush(), close() — yang dipenuhi dua implementasi: simulasi LocalProducer dan adapter KafkaProducer dari kafka-python untuk cluster KRaft sungguhan. Pabrik berikut menjadi satu-satunya titik pergantian:

python 
def buat_producer(backend: str = "local", **kwargs):
    """Pabrik producer sesuai protokol (``send``/``flush``/``close``).

    ``backend="local"`` mengembalikan :class:`LocalProducer` (butuh ``broker``
    berupa :class:`~src.local_kafka.LocalKafka`). ``backend="kafka"`` memuat
    adapter kafka-python untuk cluster sungguhan; adapter tersebut tidak
    dijalankan di lingkungan pengembangan proyek ini.
    """
    if backend == "local":
        return LocalProducer(**kwargs)
    if backend == "kafka":
        from .kafka_client_adapter import buat_producer_kafka

        return buat_producer_kafka(**kwargs)
    raise ValueError(f"backend tidak dikenal: {backend!r}")

Adapter src/kafka_client_adapter.py menerjemahkan protokol yang sama ke pemanggilan kafka.KafkaProducer dengan value_serializer JSON UTF-8 dan acks="all"; berkas tersebut diverifikasi sintaks dengan py_compile dan tidak dijalankan di lingkungan ini karena pustaka kafka-python tidak terpasang. Berpindah ke cluster nyata nanti cukup mengganti satu argumen: buat_producer(backend="kafka", bootstrap_servers="localhost:9092").

send() dan flush() pada LocalProducer #

Inti producer ada di LocalProducer.send pada src/local_kafka.py:

python 
    def send(
        self,
        topic: str,
        key: str | bytes | None = None,
        value: str | bytes | None = None,
        partition: int | None = None,
        timestamp_ms: int | None = None,
    ) -> FutureRecordMetadata:
        """Buffer satu pesan; partition dipilih partitioner bila tidak ditentukan."""
        self._broker._cek_topik(topic)
        jumlah = len(self._broker.partitions_for_topic(topic))
        if partition is None:
            partition = self._partitioner.pilih(key, jumlah)
        tp = TopicPartition(topic, partition)
        future = FutureRecordMetadata()
        pesan = {
            "key": _baca_bytes(key, "key"),
            "value": _baca_bytes(value, "value"),
            "timestamp_ms": timestamp_ms,
        }
        self._buffer.append((tp, pesan, future))
        if self.linger_ms == 0:
            self.flush()
        elif sum(1 for tp2, _, _ in self._buffer if tp2 == tp) >= self.batch_size:
            self.flush()
        return future

Mari kita telaah baris-baris kode di atas (nomor mengikuti posisi di src/local_kafka.py):

  1. Baris 345 (_cek_topik): producer menolak menulis ke topic yang belum dibuat — pada simulasi, topik tidak dibuat otomatis agar kesalahan konfigurasi tampak sejak dini.
  2. Baris 347-348 (partition = self._partitioner.pilih(...)): partition dipilih partitioner hanya bila pemanggil tidak menentukannya; pesan berkunci di-hash (Bagian 5), pesan tanpa kunci bergilir.
  3. Baris 350-355 (future dan pesan): send() tidak langsung menulis; pesan masuk buffer per partition dan future dikembalikan segera — sama seperti kafka-python yang menumpuk pesan di record batch sebelum benar-benar mengirim.
  4. Baris 357-360 (flush() bersyarat): linger_ms=0 mengirim langsung (perilaku bawaan); linger_ms>0 menahan pesan sampai flush() dipanggil eksplisit atau buffer partition mencapai batch_size — model sederhana dari linger.ms dan batch.size yang menukar sedikit latensi dengan throughput.
  5. Baris 361 (return future): nasib pesan sesungguhnya dibaca belakangan lewat future.get(), yang baru terisi setelah flush() menulis ke log broker.

Simulator Checkout #

Halaman checkout web tidak tahu apa-apa soal Kafka; kode aplikasi cukup memanggil simulasi berikut. Kelas ini juga menjaga penomoran event antar panggilan sehingga hasil produksi bisa diperpanjang tanpa menimpa:

python 
    def kirim(self, jumlah: int) -> list[RecordMetadata]:
        """Kirim ``jumlah`` event baru; nomor event berlanjut antar panggilan."""
        events = buat_pesanan_events(
            jumlah, seed=self._seed, mulai_nomor=self._nomor + 1,
            bobot_wilayah=self._bobot,
        )
        self._nomor += jumlah
        return self.kirim_events(events)

    def kirim_events(self, events: list[dict]) -> list[RecordMetadata]:
        """Kirim daftar event apa adanya; kembalikan metadata tiap pesan."""
        futures = [
            self._producer.send(
                self.topic,
                key=event.get(self._kunci),
                value=serialisasi_event(event),
            )
            for event in events
        ]
        self._producer.flush()
        return [future.get() for future in futures]

kirim_events memperlihatkan pola producer yang benar: seluruh send() ditumpuk lebih dulu, flush() mengirim semuanya, dan baru setelah itu future.get() diambil — memanggil get() sebelum flush akan menolak karena pesan masih di buffer (perilaku yang dikunci tes test_producer_send_flush_dan_future).

Menjalankan Producer #

Simulasi berikut mengirim 10 event pesanan berkunci pelanggan_id lalu memeriksa isi partition:

python 
import json

from src.local_kafka import LocalKafka
from src.producer import SimulatorCheckout, buat_producer

broker = LocalKafka()
broker.create_topic("tokokita.pesanan", num_partitions=3)

producer = buat_producer(broker=broker, client_id="checkout-web")
checkout = SimulatorCheckout(producer, topic="tokokita.pesanan", kunci="pelanggan_id")
metadatas = checkout.kirim(10)
for m in metadatas[:4]:
    print(f"terkirim: topic={m.topic} partition={m.partition} offset={m.offset}")

print(f"... total {len(metadatas)} event terkirim")
print(f"describe_topic: {broker.describe_topic('tokokita.pesanan')}")

print("isi partition 0 (kunci -> event):")
for rekam in broker.baca("tokokita.pesanan", 0):
    event = json.loads(rekam.value.decode("utf-8"))
    print(f"  offset {rekam.offset}: {rekam.key.decode()} -> {event['event_id']}")

Output yang Dihasilkan:

text 
terkirim: topic=tokokita.pesanan partition=0 offset=0
terkirim: topic=tokokita.pesanan partition=0 offset=1
terkirim: topic=tokokita.pesanan partition=0 offset=2
terkirim: topic=tokokita.pesanan partition=2 offset=0
... total 10 event terkirim
describe_topic: {'topic': 'tokokita.pesanan', 'partitions': 3, 'total_pesan': 10}
isi partition 0 (kunci -> event):
  offset 0: PLG-115 -> evt-000001
  offset 1: PLG-191 -> evt-000002
  offset 2: PLG-006 -> evt-000003
  offset 3: PLG-124 -> evt-000005

Dua hal terbaca dari output: offset bertambah monoton per partition (0, 1, 2, 3 pada partition 0), dan evt-000004 tidak ada di partition 0 karena kunci pelanggannya (misalnya PLG-019) di-hash ke partition lain — bukan karena pesan hilang. Kelima pesanan partition 0 tetap terurut sesuai urutan pengiriman; jaminan urutan inilah yang membuat kunci penting, dan kita ukur konsekuensinya pada Bagian 5.

Event sudah mengalir ke topic. Sekarang kita bangun sisi pembacanya: consumer dengan poll loop yang mengagregasi jendela per menit.