Memproduksi pesan

Topik ini membahas cara producer menulis pesan ke broker Kafka, mulai dari konfigurasi dasar pada klien Python, alur pesan dari panggilan send() hingga batch terkirim, sampai serialisasi nilai pesan. Dua demo melengkapi pembahasannya, yaitu pasangan producer dan consumer dari CLI serta program Python yang menampilkan metadata pengiriman. Setelah membacanya, perilaku konfigurasi acks, linger.ms, dan batch.size dapat dijelaskan dari mekanismenya, bukan sekadar dari daftar parameter.

Konfigurasi dasar producer#

Producer minimal hanya membutuhkan alamat bootstrap server, misalnya localhost:9092 pada broker lokal, sedangkan konfigurasi lainnya mengikuti nilai default-nya. Konfigurasi yang paling menentukan perilaku pengiriman adalah tiga parameter berikut, yang mencakup tingkat konfirmasi yang ditunggu producer dan cara pesan dikumpulkan sebelum dikirim.

Konfigurasi acks menerima tiga nilai. Nilai 0 membuat producer tidak menunggu konfirmasi apa pun, sehingga pengiriman paling cepat tetapi paling rentan hilang. Nilai 1 menunggu konfirmasi leader partisi saja, sedangkan nilai all menunggu konfirmasi seluruh salinan yang sinkron sehingga paling tahan terhadap kegagalan broker. Nilai default all berlaku bersama idempotence, yaitu jaminan bebas duplikasi yang juga aktif secara default pada producer modern.

Konfigurasi linger.ms dan batch.size bekerja berpasangan pada sisi efisiensi. Producer menunggu paling lama linger.ms sebelum menyerahkan satu batch, tetapi menyerahkannya lebih cepat apabila batch sudah mencapai batch.size byte. Jeda kecil ini mengurangi jumlah permintaan jaringan karena beberapa pesan berangkat bersama, dengan konsekuensi latensi paling banyak setara jeda tersebut. Sejak Kafka 4.0 nilai default linger.ms berubah dari 0 menjadi 5 demi efisiensi pada beban kerja umum.

Alur send sampai batch terkirim#

Metode send() tidak langsung mengirim pesan ke jaringan. Panggilan itu menjalankan serialisasi nilai dan key, memilih partisi tujuan, lalu menyisipkan pesan ke record accumulator, yaitu penyangga yang menampung batch per partisi. Satu thread terpisah bernama sender mengambil batch yang siap dari penyangga dan menyampaikannya ke broker leader partisi tersebut. Setelah broker menambahkan pesan ke log dan mengirim konfirmasi sesuai nilai acks, pengiriman itu dianggap selesai.

Sebagai balasannya, send() mengembalikan objek future yang berisi metadata pengiriman, yaitu nomor partisi dan offset akhir pesan. Metode get() pada future itu menunggu satu pesan tertentu selesai, sedangkan flush() menunggu seluruh pesan yang masih di penyangga, dan close() menutup producer secara rapi sambil menyelesaikan sisa pengiriman.

Producer dan consumer dari CLI#

Alur dasar pengiriman dapat diuji tanpa program apa pun menggunakan producer dan consumer console. Producer console membaca setiap baris dari masukan standar sebagai satu pesan, sehingga printf cukup mengirim tiga baris sekaligus. Consumer console membaca dari awal log dengan --from-beginning dan berhenti sendiri setelah tiga pesan karena opsi --max-messages 3. Topic demo menggunakan nama yang diakhiri penanda waktu, sehingga setiap eksekusi memperoleh topic baru dengan log yang kosong.

(
  cd /home/kal/my-project/kafka-runtime/kafka
  T="m1-demo-aliran-$(date +%s)"
  bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic "$T" --partitions 1 >/dev/null
  sleep 2
  printf "antrean-101\nantrean-102\nantrean-103\n" | bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic "$T" 2>/dev/null
  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic "$T" --from-beginning --max-messages 3 2>/dev/null
)

antrean-101
antrean-102
antrean-103

Tiga baris yang dikirim producer kembali persis dengan urutan yang sama karena ketiganya berada di satu partisi. Consumer tidak mencetak baris ringkasan apa pun karena --max-messages mengakhirinya dengan bersih begitu kuota pesan terpenuhi. Demo ini memperlihatkan cara paling sederhana, sedangkan program Python berikut memperlihatkan metadata yang dihasilkan setiap pengiriman.

Producer Python dan serialisasi nilai#

Producer Kafka mengirim nilai dalam bentuk bytes, sehingga serialisasi dilakukan sebelum panggilan send(). Pada demo berikut, serialisasi menggunakan encode() yang mengubah string menjadi bytes UTF-8, dan konfigurasi yang dibahas di awal dipasang langsung saat KafkaProducer dibuat. Metadata setiap pengiriman dibaca dari future dengan get(), lalu seluruh pesan dibaca kembali oleh consumer untuk membuktikan bahwa semuanya tiba di log.

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

topic = f"m1-produksi-{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()

producer = KafkaProducer(
    bootstrap_servers="localhost:9092",
    acks="all",
    linger_ms=5,
    batch_size=16384,
)

for i in range(5):
    future = producer.send(topic, value=f"bacaan-{i}".encode())
    info = future.get(timeout=10)
    print(f"terkirim partisi={info.partition} offset={info.offset}")

producer.flush()
producer.close()

consumer = KafkaConsumer(
    topic,
    bootstrap_servers="localhost:9092",
    auto_offset_reset="earliest",
    consumer_timeout_ms=2000,
)
for msg in consumer:
    print(f"terima offset={msg.offset} nilai={msg.value.decode()}")
consumer.close()

terkirim partisi=0 offset=0
terkirim partisi=0 offset=1
terkirim partisi=0 offset=2
terkirim partisi=0 offset=3
terkirim partisi=0 offset=4
terima offset=0 nilai=bacaan-0
terima offset=1 nilai=bacaan-1
terima offset=2 nilai=bacaan-2
terima offset=3 nilai=bacaan-3
terima offset=4 nilai=bacaan-4

Setiap baris terkirim membuktikan bahwa future-nya sudah selesai, dengan nomor partisi dan offset yang ditetapkan broker. Pada topic baru dengan satu partisi, kelima pesan menempati offset 0 sampai 4 secara berurutan, dan consumer membacanya dengan urutan yang sama. Konfigurasi acks="all" membuat setiap panggilan get() selesai setelah konfirmasi lengkap diterima, sehingga keluaran ini juga menunjukkan pengiriman yang terjamin dari sisi klien.

Kesalahan umum#

1. Mengirim objek str tanpa serialisasi memunculkan error karena producer hanya menerima bytes. Perbaikannya dengan encode() pada setiap nilai, atau dengan mengatur value_serializer yang mengubah string menjadi bytes secara otomatis pada saat KafkaProducer dibuat.
2. Program berakhir tanpa flush() atau close() sehingga pesan-pesan terakhir masih berada di penyangga dan tidak pernah sampai ke broker. Keluaran consumer kemudian berhenti lebih pendek dari yang diharapkan. Selalu akhiri program dengan penyelesaian pengiriman sebelum prosesnya berhenti.
3. Mengirim ke topic yang belum ada memunculkan error UNKNOWN_TOPIC_OR_PARTITION ketika pembuatan topic otomatis pada broker dimatikan, keadaan yang merupakan default pada cluster produksi. Buat topic terlebih dahulu lewat KafkaAdminClient atau bin/kafka-topics.sh, lalu jalankan producer-nya.

Ringkasan#

• Panggilan send() menempatkan pesan pada penyangga batch per partisi, dan thread sender menyampaikan batch ke broker.
• Batch dikirim saat ukurannya mencapai batch.size, saat jeda linger.ms berakhir, atau saat flush() dipanggil.
• acks menetapkan konfirmasi yang ditunggu producer, dengan nilai default all yang paling tahan terhadap kegagalan broker.
• Nilai pesan wajib berupa bytes, sehingga serialisasi dilakukan sebelum pengiriman, misalnya lewat encode().

Referensi#

• Producer configs — Apache Software Foundation — daftar lengkap konfigurasi producer beserta nilai default acks, linger.ms, dan batch.size.
• Apache Kafka documentation — Apache Software Foundation — dokumentasi utama Kafka yang mencakup model pengiriman pesan dari producer ke broker.
• kafka-python KafkaProducer — kafka-python project — dokumentasi kelas KafkaProducer yang digunakan pada demo Python topik ini.