ما تا الان نحوه کار کلی کافکا، دموی کوچیکی از اون و نصبش رو با هم مشاهده کردیم. در این نوشتار میخوایم موارد پیشرفته ی کافکا که شامل متادیتا و جزییات پیام ها، همچنین ذخیره سازی و دریافت پیغامها توسط کافکا رو بیشتر بدونیم. این جزییات برای مصارفی که از دست نرفتن حتی یک پیغام هم حیاتی است بسیار لازم است.
در کافکا بستر متادیتا با استفاده از سه بخش key ،value و header فراهم شده است. بخش header اجباری است ولی key، value ها اختیاری هستند. اغلب توصیه میشود مستقیم درگیر هدرها و ارتباط با درایور کافکا نشوید. ولی برای مثال در کاربردهایی مانند audit پیام(پیام رسیده باشد و state کنونی در سامانه چیست) و یا track پیام ها، از هدر استفاده میشود. برای مثال id پیام را در هدر قرار میدهیم و به راحتی track پیام را در جاهای مختلف داشت. در مبحث پیام رسان ها هم اغلب ادعا میشود ما end to end encryption داریم و اگر نخواهند درگیر verify کردن sign با باز کردن پیام شوند میتوانند اطلاعاتی را با استفاده از هدرها منتقل کنند.
کافکا به صورت دیفالت تایپ های String، Double، Long، Integer و Byte رو خودش serialize و deserialize میکند. اما در داخل خود پلتفرم چه در مورد متا دیتا و چه پیغام ها، تایپ داده به byte array تبدیل میشود. اگه بخوایم دستی این سریالایز، دی سریالایز رو انجام ندیم باید سریالایزر custom بنویسیم. سریالایزر تعریف کردن یه کار اضافه ای هم انجام میده که ما اسکیما رو دستی چک نمیکنیم و خود پلتفرم بررسی میکنه که داده ها اون تایپ رو داشته باشن. اگر یه وقت بخوایم compression داشته باشیم یا sign و encryption انجام بدیم باید در بخش سریالایزیشن این رو ببینیم.
ما تا اینجا تنها یک بروکر استفاده کردیم. در صورتی که این بروکر به هر علتی مشکلی داشته باشد producer ها دیگر قادر به ارسال پیام نیستند و پیام های ارسالی تا کنون نیز اگر به دست consumer ها نرسیده باشد از بین رفته است. بنابراین در صورت داشتن تک بروکر، در مقابل مشکل خرابی بروکر تحمل پذیر نیستیم و در صورت داشتن تنها یک بروکر، لازم است بهای از دست رفتن پیغام ها و downtime را بپردازیم. برای پیش نیامدن اینطور مشکلاتی، کافکا راه حل استفاده از چند بروکر را دارد. مدیریت وجود چند بروکر در کافکا را zookeeper بر عهده دارد. بروکر ها خود را در zookeeper رجیستر کرده و از طریق آن وجود باقی بروکر ها را متوجه میشوند. این تنها کاری است که zookeeper در کافکا انجام میدهد.
در صورتی که ما تنها یک بروکر داشته باشیم یا داده ها را تنها در یک بروکر ذخیره کنیم، ممکن است data loss اتفاق بیافتد. برای همین کافکا ویژگی دیگری به نام replication factor را ارائه داده است.برای جلوگیری از fail نشدن و از دست نرفتن پیغام زمانی که broker غیرفعال شود، هر پیغام به تعدادی broker همزمان ارسال میشود. به این تعداد کپی از داده ها replication factor گفته میشود.(اصل داده به بروکر leader میرسد و از آنجا در سایر بروکر ها ذخیره میشود.) در زمان تعریف topic، مقدار replication مورد نیاز نیز قابل تنظیم است. با تنظیم این عدد به این تعداد کپی بر روی بروکر های متفاوت از این داده ذخیره میشوند. بنابراین لازم است همزمان به تعداد replication factor بروکر در دسترسی باشد. داده ها بر روی چندین بروکر ذخیره میشود ولی producer تنها با یک بروکر در ارتباط است که به آن partition leader میگویند.
نحوه انتخاب این election به این صورت است که zookeeper از بروکرها میپرسد چه کسی حاضر است لیدر باشد و همه بروکرها درخواست لیدر شدن خود را ارسال میکنند و هر کدام از درخواست ها که سریع تر به zookeeper برسد به عنوان لیدر انتخاب میشود!
در ابتدای کار با کافکا یک سری configuration تنظیم میکنیم و برنامه برای تولید/مصرف پیغام بالا می آید، پیش از ارسال داده یا subscribe شدن consumer، یک meta data رد و بدل میشود. در این متادیتا تعداد replication ها، protocol تبادلی، زمان های insync، لگ، leader هر پارتیشن ، تعداد پارتیشن ها و ... مشخص میشود. بنابراین در ابتدا producer و consumer میداند leader کیست. پیغام تنها برای leader ارسال میشود و این بروکرهای داخل کلاستر هستند که با هم تعامل دارند تا داده را بین خود نیز تبادل کنند. بروکر ها به partition leader درخواست یا issue ای برای fetch data ارسال میکند تا کپی ای از آن partition داشته باشند. کافکا لازم است به نحوی بفهمد که پراسس داده ها تمام شده است، اینکار با دریافت پیغام offset commit انجام میشود.
در صورتی که leader از دست برود، یک تاخیر چند میلی ثانیه ای ایجاد میشود و بعد از آن بین بروکر ها یک election (مثل دفعه قبل) انجام میشود و لیدر بعدی انتخاب میشود و دوباره این leader در غالب متا دیتا به همه معرفی میشود. isr (In Sync Replica) تعداد بروکر هایی که داده هایشان با هم synch میباشد است. این بروکر ها با یک lag ای سینک میشوند که قابل تنظیم میباشد. (replica lag time max ms) پیش فرض این لگ تایم 30 ثانیه میباشد. leader ها همواره لیستی از بروکرهای in sync replica را دارند. هر بروکر ماکسیمم طی بازه لگ تایم باید sync شود و اگر بیشتر از این خود را آپدیت نکرده باشد، out of sync در نظر گرفته میشود.
در کافکا مسئله دیگری که مهم است delivery و durability است. در مبحث durability باید براساس بیزینس و مسئله تصمیم گیری کرد. گاهی مهم است که تمامی پیام هایی که ارسال میشوند، پردازش شود. ولی در بعضی مسائل اگر 90 درصد پیغام ها پردازش شوند نیز کفایت میکند و از دست دادن 10 درصد باقی مانده مشکلی ایجاد نمیکند. بنابراین بیزینس مسئله تعیین کننده ی durability است. پیغام از جانب producer تولید و در بروکر قرار میگیرد. بروکر تضمین میکند که براساس کانفیگ های انجام شده، پیغام به دست consumer میرسد. این تضمین با کمک acks (acknowledge) انجام میشود. acks سه مقدار متفاوت none، one و all دارد. برای مثال در صورتی که بیزینس track رفتار کاربران باشد دریافت تک تک این کلیک ها کم اهمیت است ولی اگر کافکا را برای transaction های بانکی به کار ببریم اطمینان از دریافت آنها بسیار مهم است. برای همین در producer متغیری به نام acks وجود دارد که در صورتی که "acks" = none، از جانب بروکر ack ای مبنی بر دریافت پیغام به producer ارسال نمیکند و producer هر پیغامی که ارسال میکند را دریافت شده فرض میکند حتی اگر بروکر ای برای دریافت نباشد. بنابراین در چنین حالتی ممکن است data loss اتفاق بیافتد.
اگر مقدار acks = 1 باشد، بروکر با دریافت پیغام یک ack به producer ارسال میکند. بنابراین در این حالت producer متوجه میشود که پیغام به درستی دریافت شده است. به محض دریافت و ذخیره ی پیغام توسط یکی از بروکرها، بروکر ack را به producer ارسال میکند ولی هنوز commit فرض نشده است. بعد از نوشته شدن داده توسط تمامی بروکرهای دیگر داده commit شده فرض میشود.
در این حالت پیش از commit بودن و ذخیره ی داده در تمامی بروکرها، ممکن است داده در partition leader ذخیره شود (ولی هنوز در سایر بروکرها ذخیره نشده باشد) و با از دست رفتن بروکر، داده نیز از دست میرود. برای جلوگیری از چنین مشکلی، acks = all میشود و مادامی که همه ی بروکرها داده را ذخیره نکرده باشند، Producer پیغام ACK را دریافت نمیکند. بنابراین اگر مقدار acks = all باشد، تمامی بروکر هایی که وجود دارند باید تضمین کنند که داده را دریافت و ذخیره کرده اند.
در حالت acks = all کافکا برای ارسال ack، متغیر دیگری را نیز لحاظ میکند. این متغیر، min.insynch.replica است. به صورت دیفالت این مقدار 1 است. کافکا در حالت acks = all رفتار دیگری نیز دارد، برای مثال یکی از بروکر ها ارتباطش را از دست داده است و داده ها را ذخیره نمیکند. اگر min.insynch.replica کمتر از replication factor باشد، کپی داده به تعداد min برای ارسال ack کفایت میکند. این رفتار عجیب کافکا برای اینکه در خیلی از مواقع ارتباط نتورکی بروکر ها از دست میرود کارایی دارد که تولید پیغام را متوقف نکند.
در صورتی که به اندازه ی min.insynch.replica بروکر فعال وجود نداشته باشد، producer پیغام خطا دریافت میکند.
در این پست به موارد و جزییات بیشتری از کافکا اشاره کردیم و برخی کانفیگ های مهم برای ارسال بدون خطای داده ها را بررسی کردیم. نحوه ی نگهداری داده ها بررسی شد و نحوه ی کانفیگ چند بروکر ار دیدیم. برای یک برنامه نویس دانستن این حدود از کافکا کفایت میکند و برخی مطالب اشاره نشده در این مطالب مگر در موارد خاص یا به عنوان admin کافکا برایتان مفید باشد.
از اینکه وقت خود را در اختیار من قرار دادین بسیار سپاس گذارم. امیدوارم این مطالب برای شما موثر و مفید باشد. من تلاش کردم مطلب رو به ساده ترین حالت بیان کنم. در صورتی که پیشنهاد یا انتقادی در مورد نحوه ی نگارش و مطالب دارین خوشحال میشم با من در میان بذارین و هر سوال و ابهامی در مورد کافکا داشتین رو در کامنت ها بذارین که بتونم کمک کنم.
