Week 1 · Day 2 · Producer internals

Batching: apa yang block, apa yang throw?

Bila link WAN dari edge ke Confluent Cloud putus, producer ko tak terus fail. Dia simpan dulu data tu. Lesson ni pasal apa jadi bila tempat simpan tu dah penuh.

Teka dulu (30 saat): link edge → broker putus 3 minit. Apa jadi kat producer.produce() dalam clients/producer.py ko? Tulis jawapan kat kertas dulu, baru scroll. Teka salah pun takpe, otak jadi lagi senang ingat lepas ni.

1. Bila batch keluar?

Recap Day 1: send() cuma append record masuk satu batch per partition dalam RecordAccumulator. Yang betul² hantar ke broker ialah sender thread, kemudian. Batch tu dikira ready bila jadi salah satu dari 3 ni:

Penuh: record seterusnya dah tak muat dalam batch.size (default 16384 bytes).
Dah tunggu lama: batch dah tunggu linger.ms (default 5 dalam Kafka 4.x; versi lama default 0).
Memori habis: ada send() lain tengah stuck tunggu memori, so semua batch terus dihantar.

Sumber: RecordAccumulator.batchReady() (full, expired, exhausted) dan defaults dalam ProducerConfig.

Pasal compression.type pulak: dia compress record sebelum masuk kira dalam batch. So lagi banyak record boleh muat dalam batch.size yang sama. Kesannya, batch jarang keluar sbb penuh, lagi selalu keluar sbb linger.ms dah habis. (saiz record dianggar ikut compression.type)

2. Memori terhad: Java block dulu, lepas tu fail ikut callback

Semua batch dalam Java producer ambil memori dari satu kolam, saiz dia buffer.memory (default 32 MB). Kalau broker tak ambik batch, memori tu tak dipulangkan. Lama-lama, kolam tu kering.

Masa tunggu metadata pun ditolak dari bajet max.block.ms yang sama tau.

Ok ni part yang ramai kena tipu. Bila send() fail, ada 2 jalan error tu sampai kat ko:

Jalan A: throw. Exception tercampak keluar dari send() tu sendiri. try/catch keliling send() boleh tangkap.
Jalan B: callback + Future. send() return macam biasa, takde apa². Error tu dihantar ke callback ko, atau keluar bila panggil future.get().

Rule dia simple je, ikut jenis exception. Kalau ApiException, maksudnya "gagal hantar record ni", so dia ikut jalan B. KafkaException lain pulak maksudnya "program ko ada bug", so dia ikut jalan A. BufferExhaustedException tu subclass TimeoutException, dan TimeoutException ialah sejenis ApiException, so dia tak di-throw dari send().

Sumber: BufferPool.allocate() · KafkaProducer.doSend() catch blocks · max.block.ms docs

Simptom dalam app Java	Punca	Macam mana dia muncul
`send()` lambat return	tunggu metadata atau memori	block, paling lama `max.block.ms`
Callback dapat `TimeoutException`	memori masih penuh lepas `max.block.ms`	jalan B: callback + Future
Callback dapat `RecordTooLargeException`	record > `max.request.size` atau > `buffer.memory`	jalan B, terus serta-merta
`send()` sendiri campak exception	serializer fail, thread kena interrupt, producer dah closed/fatal	jalan A: throw (`SerializationException`, `InterruptException`, `KafkaException`)

Kenapa SRE kena peduli: kalau app panggil send() tanpa callback dan takde siapa check Future, semua error jalan B hilang senyap. Takde log, takde alert. Rule of thumb: jgn skip callback.

3. Producer edge ko (librdkafka) lain perangai

Client Python dalam lab guna librdkafka. Dia takde buffer.memory, takde max.block.ms. Ganti dia ada satu queue local, dihadkan oleh queue.buffering.max.messages (default 100000) dan queue.buffering.max.kbytes (default 1048576 KB). Bila queue penuh, produce() terus raise BufferError. Takde tunggu langsung.

Sumber: librdkafka CONFIGURATION.md · Producer.produce: "Raises BufferError … queue.buffering.max.messages exceeded"

Ni apa yang clients/producer.py ko buat bila kena BufferError:

except BufferError:
    # Local queue full: wait for deliveries, then retry
    producer.poll(0.1)

Plot twist: code ni sebenarnya buat "block" sendiri, dan takde had masa langsung. Dalam masa yang sama, setiap mesej dalam queue ada jam sendiri. Dalam edge-producer.properties ko, delivery.timeout.ms=120000. Bila mesej dah expired, dia dilaporkan fail melalui on_delivery, dan tempat dia dalam queue dikosongkan.

4. Latihan

Jawab ikut ingatan. Jgn scroll atas dulu, no cheating 👀

5. Lab 10 minit 💻 desktop only

Guna lab 3-broker, BUKAN node edge production.

Ni preview weekend lab #3. Tulis dulu ramalan ko untuk 2 nombor: saat ke berapa BufferError start, dan saat ke berapa delivery mula fail. Lepas tu run:

cd ~/workspaces/kafka-in-depth  # folder ni
python clients/producer.py -t orders -n 0 -r 2000 -s 1000 -q \
  -X queue.buffering.max.messages=20000 -X delivery.timeout.ms=30000
# kat terminal lain, lepas ~5 s:
podman stop kafka-1 kafka-2 kafka-3
# lepas ~45 s, Ctrl-C producer, baca summary, lepas tu:
podman start kafka-1 kafka-2 kafka-3

Check ramalan

Queue penuh lebih kurang 10 s lepas broker stop (20000 ÷ 2000). Lepas tu producer.py berpusing dalam loop poll(0.1) diam² tanpa print apa². Kalau nak nampak, tambah print sekejap dalam blok except BufferError. Delivery mula fail lebih kurang 30 s lepas mesej pertama yang tak terhantar dibuat, dan summary lepas Ctrl-C patut tunjuk failed count yang bukan kosong. Kalau nombor ko lari jauh, bawak output tu sini, kita bedah sama².

Bacaan utama (15 minit): KafkaProducer.doSend() dalam Kafka 4.1. Lebih kurang 100 baris je, dan inilah yang tentukan semua baris dalam table atas: tunggu metadata, serialize, partition, append, dan catch mana yang throw balik vs mana yang bagi kat callback.
Cheat sheet: Producer batching: Java vs librdkafka.

Ada yang blur? Tanya je terus dalam chat. Contohnya "kenapa tunggu metadata pun kira dalam max.block.ms?" atau "camne nak sizing queue edge untuk outage 1 jam?" Aku cikgu ko untuk topik ni.