Producer در Kafka: معماری، پیکربندی و جریان دقیق پیام‌ها

Producerها همان نقطه ورود داده به Kafka هستند. وظیفه اصلی آنها ارسال پیام‌ها به پارتیشن‌های مناسب و لیدر مربوطه است. اما فرآیند ساده “ارسال یک پیام” پشت پرده شامل چندین مرحله مهم است:

1. دریافت metadata هر پارتیشن و لیدرها از طریق اتصال به یکی از brokerهای bootstrap.
2. تعیین پارتیشن مقصد برای هر پیام، بر اساس key یا الگوریتم round-robin.
3. ارسال پیام‌ها به صورت batch برای بهینه‌سازی throughput و کاهش بار شبکه.
4. دریافت تاییدیه (ack) از لیدر، با توجه به سطح دوام مورد نیاز (acks).
5. مدیریت تکرار پیام‌ها و اطمینان از exactly-once delivery با idempotency یا تراکنش‌ها.

Producer علاوه بر ارسال، مسئول مدیریت حافظه محلی، batching، retry و ترتیب پیام‌ها نیز هست.

🧭 مرحله اول: آغاز کار Producer و Metadata Discovery

قبل از ارسال اولین پیام، Producer نیاز دارد بداند:

• هر تاپیک چه پارتیشن‌هایی دارد
• هر پارتیشن چه لیدری دارد
• کدام broker کنترلر اصلی است

این اطلاعات metadata در cache Producer ذخیره می‌شود و هر چند دقیقه یکبار یا در صورت خطا، بروزرسانی می‌گردد.

جریان عملی

۱. اتصال به یکی از bootstrap brokers
۲. ارسال MetadataRequest
۳. دریافت:
    - لیست brokerها
    - Controller ID
    - Topic → Partition → Leader mapping
۴. ذخیره metadata محلی

بدون این مرحله، Producer نمی‌تواند تصمیم بگیرد که پیام را به کدام پارتیشن و لیدر ارسال کند.

⚙️ مرحله دوم: فراخوانی .send() و آماده‌سازی Batch

هنگامی که Producer پیامی را برای ارسال، دریافت می‌کند:

1. پیام serializable می‌شود (key و value به bytes تبدیل می‌شوند).
2. پیام در RecordAccumulator برای هر (topic, partition) ذخیره می‌شود.
3. زمانی که batch پر شد یا زمان linger.ms رسید، batch به لیدر پارتیشن ارسال می‌شود.

این مرحله مهم است چون باعث افزایش throughput و کاهش تعداد درخواست‌های شبکه می‌شود.

🔁 سطح دوام پیام: acks

Producer می‌تواند تعیین کند که چه تعداد ack از Broker قبل از موفقیت ارسال نیاز است.

• چرا لازم است؟
  بدون کنترل ack، پیام ممکن است از دست برود یا تکراری شود، مخصوصاً در زمان crash یا failover لیدر.
• پارامتر متناظر برای این موضوع در تولیدکننده: acks

این پارامتر مستقیماً بر latency و durability پیام تاثیر می‌گذارد.

🧩 چرا به Idempotency نیاز داریم؟

در سناریوهای واقعی، Producer ممکن است به دلیل خطا یا timeout، پیام‌ها را دوباره ارسال کند. بدون idempotency، این کار باعث تکرار پیام‌ها در log پارتیشن می‌شود.

• راهکار: فعال کردن enable.idempotence=True
• چطور کار می‌کند:
  1. Broker به Producer یک PID (Producer ID) یکتا می‌دهد.
  2. Producer برای هر (topic, partition) sequence number نگه می‌دارد.
  3. هر batch با PID و sequence ارسال می‌شود.
  4. Broker بررسی می‌کند:
     • Duplicate → discard
     • Next sequence → append
     • Out-of-order → خطا

نتیجه: حتی با retry یا crash، هر پیام فقط یکبار ذخیره می‌شود.

🔐 تراکنش‌ها و Exactly-Once Across Partitions

گاهی نیاز است که پیام‌ها چند پارتیشن یا چند تاپیک را به صورت atomic ارسال کنند.( این موضوع را با جزییات بیشتر در ادامه دوره بررسی می کنیم )

• نیاز عملی: جلوگیری از partial write و inconsistent state در سیستم‌های downstream.
• راهکار Producer: transactional.id + enable.idempotence=True

جریان تراکنش

init_transactions()
begin_transaction()
produce(... multiple topics)
commit_transaction() or abort_transaction()

تمام پیام‌ها یا commit می‌شوند یا abort، تضمین می‌کند که هیچ داده نیمه‌کاره‌ای در کلاستر نماند.

⚙️ مدیریت حافظه و Batch

Producer از حافظه محلی (buffer.memory) برای نگه‌داری پیام‌ها تا زمان ارسال استفاده می‌کند.

• max.block.ms: حداکثر زمانی که send() منتظر می‌ماند اگر حافظه پر باشد.
• batch.size / linger.ms: کنترل می‌کند که پیام‌ها چه زمانی و به چه اندازه به صورت batch ارسال شوند.

این پارامترها Tradeoff بین Latency و Throughput را مدیریت می‌کنند.

🔑 Key و Partitioning

• اگر key ارائه شود: پیام به پارتیشن مشخص (hash(key)) می‌رود → ترتیب پیام برای آن key حفظ می‌شود.
• اگر key ارائه نشود: round-robin → توزیع یکنواخت.

این تصمیم باعث می‌شود ترتیب پیام برای entity-specific یا session-specific حفظ شود.

💬 خلاصه جریان Producer

۱. Producer Start
   → Connect bootstrap servers
   → Fetch metadata
   → Assign PID (if idempotent)
   → Prepare RecordAccumulator

۲. Send()
   → Serialize key/value
   → Append to per-partition batch
   → Flush batch when batch.size پر شد یا linger.timeout
   → Send to partition leader
   → Leader assigns offset, writes to log
   → Replicas fetch and ack
   → Producer receives ack (based on acks)
   → If idempotent, update (PID, sequence)

🧰 جدول پارامترهای کلیدی Producer (با توضیحات فنی)