حالت تاریک
فصل ۷ — Topicها و پارتیشنها
این فصل شامل:
- پارامترهای ایجاد و گزینههای پیکربندی
- نحوهٔ وجود پارتیشنها بهصورت فایلهای log
- تأثیر segmentها بر داده داخل پارتیشنها
- تست با EmbeddedKafkaCluster
- فشردهسازی topic و نحوهٔ نگهداری داده
در این فصل، بیشتر به این میپردازیم که چگونه میتوانیم دادههایمان را در topicها ذخیره کنیم و همچنین چگونه topicها را ایجاد و نگهداری کنیم. این شامل نحوهٔ جایگیری پارتیشنها در ملاحظات طراحی ما و نحوهٔ مشاهدهٔ داده روی brokerهاست. همهٔ این اطلاعات به ما کمک میکند هنگام بررسی نحوهٔ بهروزرسانی داده در یک topic بهجای append کردن آن به log، بهتر عمل کنیم.
۷.۱ Topicها
برای تازهسازی حافظه، مهم است بدانیم topic یک مفهوم غیرملموس است، نه ساختار فیزیکی. معمولاً فقط روی یک broker وجود ندارد. بیشتر برنامههایی که دادهٔ Kafka را مصرف میکنند، داده را در یک topic واحد میبینند؛ برای subscribe کردن به جزئیات دیگر نیازی ندارند.
با این حال، پشت نام topic یک یا چند پارتیشن وجود دارد که واقعاً داده را نگه میدارند [1]. Kafka دادهای که یک topic را در cluster تشکیل میدهد در logها مینویسد که روی filesystem brokerها نوشته میشوند.
شکل ۷-۱. نمونهٔ topic با پارتیشنها
The topic kinaction_helloworld is made up of three partitions that will likely be spread out among different brokers. Topic: kinaction_helloworld A single partition's copy is not split between brokers and has a physical footprint on each disk. Partition 0 Partition 1 Partition 2 Kafka writes the data that makes up a topic in the cluster to logs, which are written to the broker filesystems. Figure 7.1 also shows how those partitions are made up of messages that are sent to the topic. Brokers
شکل ۷-۱ پارتیشنهایی را نشان میدهد که topicی به نام kinaction_helloworld را تشکیل میدهند. topic kinaction_helloworld از سه پارتیشن تشکیل شده که احتمالاً بین brokerهای مختلف پخش شدهاند. کپی یک پارتیشن واحد بین brokerها تقسیم نمیشود و روی هر دیسک اثر فیزیکی دارد. این شکل همچنین نشان میدهد آن پارتیشنها از پیامهایی تشکیل شدهاند که به topic ارسال میشوند.
اگر نوشتن روی topic در مثالهای شروع کار اینقدر ساده است، چرا باید نقش و اجزای تشکیلدهندهٔ topic را بفهمیم؟ در بالاترین سطح، این بر نحوهٔ دسترسی consumerها به داده تأثیر میگذارد.
فرض کنید شرکت ما با یک برنامهٔ وب-based جا برای کلاس آموزشی میفروشد که رویدادهای اقدامات کاربر را به cluster Kafka میفرستد. فرآیند کلی برنامهٔ ما میتواند حجم عظیمی رویداد تولید کند. برای مثال، رویدادی برای جستجوی اولیهٔ مکان وجود دارد، یکی برای آموزش خاصی که مشتری انتخاب میکند، و سومی برای کلاسهایی که تأیید شدهاند. آیا برنامههای producing باید همهٔ این دادهها را به یک topic واحد بفرستند یا چند topic؟ آیا هر پیام نوع خاصی از رویداد است و باید در topicهای مختلف جدا بماند؟ با هر رویکرد تنظیماتی وجود دارد و نکاتی هست که در هر موقعیت به تعیین بهترین روش کمک میکند.
طراحی topic را دو مرحلهای میبینیم. مرحلهٔ اول به رویدادهایی که داریم نگاه میکند. آیا در یک topic جا میگیرند یا بیش از یک؟ مرحلهٔ دوم هر topic را در نظر میگیرد. تعداد پارتیشنهایی که باید استفاده کنیم چقدر است؟ بزرگترین takeaway این است که پارتیشنها سؤال طراحی per-topic هستند و محدودیت یا الزام cluster-wide نیستند. اگرچه میتوان تعداد پیشفرض پارتیشن برای ایجاد topic را تنظیم کرد، در بیشتر موارد باید در نظر بگیریم topic چگونه استفاده میشود و چه دادهای نگه میدارد.
باید دلیل محکمی برای انتخاب تعداد مشخصی پارتیشن داشته باشیم. Jun Rao مقالهٔ فوقالعادهای با عنوان «How to choose the number of topics/partitions in a Kafka cluster?» را در وبلاگ Confluent دربارهٔ همین موضوع نوشته [2]! فرض کنید بهعنوان قاعدهٔ کلی میخواهیم برای هر سرور یک پارتیشن داشته باشیم. با این حال، چون یک پارتیشن روی هر سرور داریم به این معنا نیست که producerها بهطور یکنواخت بین آنها مینویسند. برای این کار باید مطمئن شویم leader هر پارتیشن به آن شکل spread out شده و همانطور بماند.
همچنین باید با دادههایمان آشنا شویم. بیایید به فهرستی از مواردی که باید دربارهٔ آنها فکر کنیم نگاه کنیم — هم بهطور کلی و هم در سناریوی کلاس آموزشی این مثال:
- صحت داده (Data correctness)
- حجم پیامهای مورد علاقه به ازای هر consumer
- چقدر داده خواهید داشت یا باید پردازش کنید
صحت داده در بالای بیشتر نگرانیهای داده در طراحیهای دنیای واقعی است. این اصطلاح ممکن است مبهم به نظر برسد؛ بنابراین تعریف ما اینجا بهعنوان نظر خودمان توضیح داده میشود. دربارهٔ topicها، این شامل اطمینان از این است که رویدادهایی که باید مرتب باشند در همان پارتیشن و در نتیجه همان topic قرار گیرند. اگرچه میتوانیم رویدادها را در consumer بر اساس timestamp مرتب کنیم، هماهنگی رویداد بین topicها در نظر ما دردسر بیشتری (و مستعد خطا) دارد تا ارزشش را داشته باشد. اگر از پیامهای keyed استفاده میکنیم و به ترتیب نیاز داریم، باید به پارتیشنها و هر تغییر آینده در آنها اهمیت دهیم [1].
برای صحت داده با سه رویداد مثال قبلی، ممکن است مفید باشد رویدادهای رزرو و تأیید/صورتحساب واقعی را با message key (شامل student ID) در دو topic جدا قرار دهیم. این رویدادها مختص دانشجو هستند و این رویکرد کمک میکند تأیید کلاس برای همان دانشجؤ خاص رخ دهد. خود رویدادهای جستجو ممکن است برای دانشجوی خاص مهم نباشند یا نیازی به ترتیب نداشته باشند؛ مثلاً اگر تیم analytics به دنبال محبوبترین شهرهای جستجو شده باشد نه اطلاعات دانشجو.
بعد باید حجم پیامهای مورد علاقه به ازای هر consumer را در نظر بگیریم. برای سیستم آموزشی نظری، با توجه به topic placement به تعداد رویدادها نگاه کنیم. رویدادهای جستجو از بقیه بیشترند. فرض کنید مکان آموزشی نزدیک شهر بزرگ روزانه ۵۰٬۰۰۰ جستجو دارد اما فقط ظرفیت ۱۰۰ دانشجو. در بیشتر روزها ۵۰٬۰۰۰ رویداد جستجو و کمتر از ۱۰۰ رویداد رزرو واقعی تولید میشود. آیا تیم تأیید برنامهای میخواهد که به topic رویداد عمومی subscribe کند و کمتر از ۱٪ کل پیامها برایش مهم باشد؟ بیشتر وقت consumer عملاً صرف فیلتر کردن انبوه رویدادها برای پردازش تعداد کمی میشود.
نکتهٔ دیگری که باید در نظر بگیریم، مقدار دادهای است که پردازش خواهیم کرد. آیا تعداد پیامها نیاز دارد چند consumer همزمان در حال اجرا باشند تا در محدودیتهای زمانی که برنامههایمان نیاز دارند پردازش انجام شود؟ اگر چنین است، باید آگاه باشیم تعداد consumerها در یک group با پارتیشنهای topic ما محدود میشود [2]. در این نقطه ایجاد پارتیشنهای بیشتر از آنچه فکر میکنیم ممکن است نیاز داشته باشیم، آسانتر است. داشتن ظرفیت بیشتر برای رشد consumerها به ما اجازه میدهد بدون نیاز به repartitioning داده، با افزایش حجم کنار بیاییم. با این حال، مهم است بدانیم پارتیشنها منبع نامحدود و رایگان نیستند، همانطور که در مقالهٔ Rao که قبلاً اشاره کردیم گفته شد. همچنین به این معناست که brokerهای بیشتری برای migrate در صورت failure broker داشته باشیم، که میتواند دردسر بالقوهای در راه باشد.
بهتر است تعادل مناسبی پیدا کنیم و با آن سیستمهایمان را طراحی کنیم.
شکل ۷-۲. نمونهٔ طراحی topic برای رویدادهای آموزشی
Partition count Topics driven by analytics workload Search type events for search topic Search messages... Book now! Partitions driven by keys and workload Rates: Booking details Customer 1 Key Value Customer 1 Location: Book/reserve Customer 1: Customer 2: Partition X Billing Customer 3: Customer 1: Customer 2
شکل ۷-۲ نشان میدهد طراحی ما ممکن است برای سه نوع رویدادی که در سناریویمان استفاده کردیم، با دو topic مناسبتر باشد. همانطور که همیشه، الزامات یا جزئیات بیشتر میتواند پیادهسازیهای آیندهٔ ما را تغییر دهد.
آخرین نکتهای که هنگام تصمیمگیری دربارهٔ تعداد پارتیشنها برای یک topic باید در نظر بگیریم این است که کاهش آن عدد فعلاً پشتیبانی نمیشود [3]. شاید راههایی برای انجام این کار وجود داشته باشد، اما قطعاً توصیه نمیشود! بیایید لحظهای فکر کنیم چرا این کار مطلوب نیست.
وقتی consumerها به یک topic subscribe میکنند، در واقع به یک پارتیشن متصل هستند. حذف یک پارتیشن میتواند موقعیت فعلی آن را از دست بدهد وقتی یا اگر consumer شروع به خواندن از پارتیشن reassigned کند. اینجاست که باید مطمئن شویم پیامهای keyed و clientهای consuming میتوانند هر تغییری که در سطح broker ایجاد میکنیم را دنبال کنند. با اقداماتمان بر consumerها تأثیر میگذاریم. حالا که طراحی topic را بحث کردیم، کمی عمیقتر به گزینههایی که میتوانیم هنگام ایجاد topic تنظیم کنیم میپردازیم. وقتی در فصل ۳ topicها را برای produce پیام ایجاد کردیم بهاختصار به اینها اشاره کردیم؛ اینجا کمی عمیقتر میرویم.
۷.۱.۱ گزینههای ایجاد topic
topicهای Kafka چند گزینهٔ اصلی دارند که برای ایجاد topic باید تنظیم شوند. اگرچه از فصل ۲ topic ایجاد کردیم (با topic kinaction_helloworld)، باید پارامترهای پایهای که glossed over شدند را بررسی کنیم. برای این پارامترها بهتر است این تصمیمات را با فکر و دقت و عمدی بگیریم [4].
تصمیم مهم دیگر در زمان ایجاد این است که آیا هرگز نیاز به حذف topic خواهید داشت. چون این عملیات مهم است، نمیخواهیم بدون تأیید منطقی رخ دهد. برای این کار Kafka نیاز دارد گزینهٔ delete.topic.enable را فعال کنیم. اگر این گزینه روی true تنظیم شود، میتوانیم topic را با موفقیت حذف کنیم و سپس حذف میشود [5].
خوب است بدانیم اسکریپتهای Kafka بهطور کلی مستندات usage خوبی دارند. پیشنهاد میکنیم ابتدا دستور kafka-topics.sh را اجرا کنید تا ببینید چه actionهای مختلفی میتوانید امتحان کنید. لیستینگ ۷-۱ دستور ناقص برای دریافت help را نشان میدهد.
لیستینگ ۷-۱. فهرست گزینههای topic
bash
bin/kafka-topics.sh # Runs the generic Kafka
# topic-related commandدر خروجی که میبینیم، یک دستور مشخص برجسته است: --create. اضافه کردن آن پارامتر به ما اطلاعات بیشتری دربارهٔ action create خودش میدهد (مثلاً «Missing required argument "[topic]"»). لیستینگ ۷-۲ دستور generic کمی بیشتر ساختهشده را نشان میدهد.
لیستینگ ۷-۲. فهرست گزینههای topic با --create
bash
bin/kafka-topics.sh --create # Lists command-specific errors
# and the help documentationچرا وقت صرف صحبت دربارهٔ این مراحل کنیم، در حالی که بعضی کاربران با manual page (man) بهعنوان بخشی از کار Linux® آشنا هستند؟ حتی اگرچه Kafka دادهای دربارهٔ نحوهٔ استفاده از tooling به آن شکل ارائه نمیدهد، این دستور قبل از اینکه مجبور شوید در Google جستجو کنید در دسترس است.
وقتی نامی داریم که بیش از ۲۴۹ کاراکتر ندارد (قبلاً امتحان شده [6])، میتوانیم topic را ایجاد کنیم. برای مثالهایمان، kinaction_topicandpart را با replication factor برابر ۲ و با دو پارتیشن ایجاد میکنیم. لیستینگ ۷-۳ syntax استفاده در command prompt را نشان میدهد [3].
لیستینگ ۷-۳. ایجاد topic دیگر
bash
bin/kafka-topics.sh # Adds the create option
--create --bootstrap-server localhost:9094 \ # to our command
--topic kinaction_topicandpart \ # Names our topic
--partitions 2 \
# Ensures that we have two
--replication-factor 2 # Creates our topic # copies of our data
# with two partitionsبعد از ایجاد topic، میتوانیم آن topic را describe کنیم تا مطمئن شویم تنظیمات درست به نظر میرسند. توجه کنید در شکل ۷-۳ partition و replication factor با دستوری که تازه اجرا کردیم مطابقت دارند.
شکل ۷-۳. describe کردن topic با دو پارتیشن
$ bin/kafka-topics.sh --bootstrap-server localhost:9094 \ --describe --topic kinaction_topicandpart Topic: kinaction_topicandpart PartitionCount: 2 ReplicationFactor: 2 Topic: kinaction_topicandpart Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2 Topic: kinaction_topicandpart Partition: 1 Leader: 2 Replicas: 2,1 Isr: 2,1
بهنظر ما، گزینهٔ دیگری که خوب است در سطح broker انجام دهیم، تنظیم auto.create.topics.enable روی false است [7]. انجام این کار اطمینان میدهد topicها را عمدی ایجاد میکنیم، نه از producer که پیامی به نام topic اشتباه تایپشده میفرستد و آن topic قبل از تلاش برای ارسال پیام واقعاً وجود نداشته. اگرچه tightly coupled نیست، معمولاً producerها و consumerها باید نام topic درست محل دادهشان را بدانند. این ایجاد خودکار topic میتواند گیجکننده باشد.
اما هنگام تست و یادگیری Kafka، topicهای autocreated میتوانند مفید باشند.
برای مثال ملموس، اگر بدون وجود آن topic این دستور را اجرا کنیم:
bash
kafka-console-producer.sh --bootstrap-server localhost:9094 --topic notexistingKafka آن topic را برایمان ایجاد میکند. و اگر این را اجرا کنیم:
bash
kafka-topics.sh --bootstrap-server localhost:9094 --listاکنون آن topic را در cluster خواهیم داشت.
اگرچه معمولاً روی حذف داده از محیطهای production تمرکز نمیکنیم، با ادامهٔ کاوش خودمان در topicها ممکن است اشتباهاتی ببینیم. خوب است بدانیم در صورت نیاز واقعاً میتوانیم topic را حذف کنیم [3]. وقتی این کار را انجام میدهیم، همهٔ داده در topic حذف میشود. این کاری نیست که مگر آمادهٔ حذف دائمی آن داده باشیم انجام دهیم! لیستینگ ۷-۴ نشان میدهد چگونه از دستور kafka-topic که قبلاً استفاده کردیم، این بار برای حذف topicی به نام kinaction_topicandpart استفاده کنیم [3].
لیستینگ ۷-۴. حذف topic
bash
bin/kafka-topics.sh --delete --bootstrap-server localhost:9094
--topic kinaction_topicandpart # Removes topic
# kinaction_topicandpartتوجه کنید گزینهٔ --delete به دستور Kafka topics ما پاس داده میشود. بعد از اجرای این دستور، دیگر نمیتوانید با این topic برای دادهتان مثل قبل کار کنید.
۷.۱.۲ Replication factorها
برای اهداف عملی، باید برنامهریزی کنیم تعداد کل replica کمتر یا مساوی تعداد broker باشد. در واقع، تلاش برای ایجاد topic با تعداد replica بیشتر از تعداد کل broker منجر به خطا میشود: InvalidReplicationFactorException [8].
میتوانیم تصور کنیم چرا این خطاست. تصور کنید فقط دو broker داریم و میخواهیم سه replica از یک پارتیشن. یکی از آن replicaها روی یک broker و دو تا روی broker دیگر وجود دارد. در این حالت، اگر brokerای که دو replica را host میکرد از دست برود، فقط یک کپی از داده باقی میماند. از دست دادن چند replica داده همزمان راه ایدهآل برای recovery در برابر failure نیست.
۷.۲ پارتیشنها
از کار با دستورات Kafka در سطح (عمدتاً) topic، شروع میکنیم به نگاه عمیقتر به پارتیشنها.
از دیدگاه consumer، هر پارتیشن log immutable از پیامهاست. فقط باید رشد کند و پیامها را به data store ما append کند. اگرچه در عمل این داده برای همیشه رشد نمیکند، تصور داده بهعنوان «اضافه شدن» بهجای «تغییر in-place» مدل ذهنی خوبی برای حفظ است.
همچنین، clientهای consumer نمیتوانند مستقیماً پیام حذف کنند. این همان چیزی است که replay پیام از topic را ممکن میکند — ویژگیای که در بسیاری سناریوها به ما کمک میکند.
۷.۲.۱ محل پارتیشن
یک چیز که ممکن است مفید باشد، نگاه به نحوهٔ ذخیرهٔ داده روی brokerهایمان است. برای شروع، بیایید محل دایرکتوری log.dirs (یا log.dir) را پیدا کنیم. محل آن را میتوان با جستجوی log.dirs در فایل server.properties پیدا کرد اگر از appendix A دنبال کرده باشید. زیر آن دایرکتوری، باید بتوانیم زیرپوشههایی با نام topic و شمارهٔ پارتیشن ببینیم. اگر یکی از آن پوشهها را انتخاب کنیم و داخلش را ببینیم، چند فایل مختلف با این پسوندها خواهیم دید: .index، .log و .timeindex. شکل ۷-۴ نشان میدهد یک پارتیشن واحد (در این مورد ۱) در topic تست ما با اجرای directory listing (ls) چگونه به نظر میرسد.
شکل ۷-۴. listing دایرکتوری پارتیشن
$ ls /tmp/kafkainaction/kafka-logs-0/kinaction_topicandpart-1/ 00000000000000000000.index 00000000000000000000.log 00000000000000000000.timeindex 00000000000000000007.index 00000000000000000007.log 00000000000000000007.timeindex 00000000000000000010.index 00000000000000000010.log 00000000000000000010.timeindex leader-epoch-checkpoint
خوانندگان دقیق ممکن است فایلی به نام leader-epoch-checkpoint و شاید حتی فایلهایی با پسوند .snapshot (در بالا نشان داده نشده) را در دایرکتوری خود ببینند. فایل leader-epoch-checkpoint و فایلهای snapshot چیزهایی هستند که وقت صرف نگاه به آنها نمیکنیم.
فایلهای با پسوند .log جایی هستند که payload دادهٔ ما ذخیره میشود. اطلاعات مهم دیگر در فایل log شامل offset پیام و همچنین فیلد CreateTime است. پس چرا به فایلهای دیگر نیاز داریم؟ چون Kafka برای سرعت ساخته شده، از فایلهای .index و .timeindex برای نگهداری نگاشت بین offset منطقی پیام و موقعیت فیزیکی داخل فایل index استفاده میکند [9].
همانطور که تا اینجا نشان داده شد، پارتیشنها از فایلهای زیادی تشکیل شدهاند. در اصل، این یعنی روی دیسک فیزیکی، پارتیشن یک فایل واحد نیست بلکه به چند segment تقسیم میشود [10]. شکل ۷-۵ نشان میدهد چند segment ممکن است یک پارتیشن را تشکیل دهند.
شکل ۷-۵. segmentها یک پارتیشن را تشکیل میدهند
Partition for kinaction_topicandpart Partition made up of Segment files Segment files one to many segments kinaction_topicandpart 7.log 10.log filename lengths are 7.index 10.index shortened for this example. 7.timeindex 10.timeindex Each segment has multiple similarly named files. Related to an Current segment inactive segment files Figure 7.5 Segments make up a partition.
segment فعال (active) فایلی است که پیامهای جدید الان در آن نوشته میشوند [11]. در illustration ما، 10.log جایی است که پیامها در دایرکتوری پارتیشن نوشته میشوند. segmentهای قدیمیتر توسط Kafka به روشهای مختلفی مدیریت میشوند که segment فعال نمیشود؛ این شامل کنترل retention بر اساس اندازهٔ پیامها یا پیکربندی زمان است. این segmentهای قدیمیتر (مثل 7.log در شکل ۷-۵) میتوانند واجد شرایط topic compaction باشند که بعداً در این فصل به آن میپردازیم.
برای جمعبندی آنچه اکنون دربارهٔ segmentها میدانیم، میدانیم چرا ممکن است چند فایل با همان نام در دایرکتوری پارتیشن داشته باشیم اما با پسوند .index، .timeindex یا .log. برای مثال، اگر ۴ segment داشته باشیم، مجموعهای از ۴ فایل داریم، هر کدام با یکی از ۳ پسوند قبلی، برای جمعاً ۱۲ فایل. اگر فقط ۱ از هر پسوند فایل ببینیم، فقط ۱ segment داریم.
۷.۲.۲ مشاهدهٔ logها
بیایید سعی کنیم نگاهی به فایل log بیندازیم تا پیامهایی که تا اینجا برای topic تولید کردهایم را ببینیم. اگر آن را در text editor باز کنیم، پیامها را در قالب human-readable نمیبینیم. Confluent اسکریپتی دارد که میتوانیم برای نگاه به آن segmentهای log استفاده کنیم [12]. لیستینگ ۷-۵ نشان میدهد دستور را به awk و grep میدهیم تا segment log فایل پارتیشن ۱ topic kinaction_topicandpart را ببینیم.
لیستینگ ۷-۵. نگاه به dump یک segment log
bash
# Prints data that cannot be
bin/kafka-dump-log.sh --print-data-log \ # viewed easily with a text editor
--files /tmp/kafkainaction/kafka-logs-0/
➥ kinaction_topicandpart-1/*.log \ # Passes a file
| awk -F: '{print $NF}' | grep kinaction # to readبا استفاده از گزینهٔ --files که الزامی است، segment file را برای نگاه انتخاب کردیم. فرض کنیم دستور موفق باشد، باید لیستی از پیامها روی صفحه چاپ شود. بدون استفاده از awk و grep، offset و metadata مرتبط دیگر مثل compression codecها را هم میبینید. این قطعاً راه جالبی برای دیدن نحوهٔ قرارگیری پیامها روی broker و دادهای که Kafka دور آن پیامها نگه میدارد است. توانایی دیدن پیامهای واقعی توانمندکننده است چون واقعاً کمک میکند log در action که Kafka را هدایت میکند را ببینید.
با نگاه به شکل ۷-۶، میتوانیم payload متنی را ببینیم که کمی خواناتر از وقتی است که سعی کردیم مستقیماً فایل log را cat کنیم. مثلاً میتوانیم پیامی در segment file با payload kinaction_helloworld ببینیم. امیدواریم دادهٔ ارزشمندتری داشته باشید!
شکل ۷-۶. مشاهدهٔ یک segment log
$ bin/kafka-dump-log.sh --print-data-log \ --files /tmp/kafkainaction/kafka-logs-0/kinaction_topicandpart-1/*.log \ | awk -F: '{print $NF}' | grep kinaction kinaction_helloworld kinaction_topicandpart kinaction_alert
در مورد عدد بزرگ در نام فایل log، تصادفی نیست. نام segment باید همان first offset در آن فایل باشد.
یکی از پیامدهای توانایی دیدن این داده این است که اکنون باید نگران باشیم چه کس دیگری میتواند آن را ببیند. چون امنیت داده و access control نگرانیهای رایج با بیشتر دادههایی است که ارزش دارد، در فصل ۱۰ راههایی را که میتوانید Kafka و topicها را secure کنید بررسی خواهیم کرد.
حقایق دربارهٔ segment log و فایلهای index جزئیاتی هستند که معمولاً در برنامههایمان به آنها تکیه نمیکنیم. با این حال، دانستن نحوهٔ نگاه به این logها ممکن است هنگام فهم نحوهٔ وجود واقعی logهایمان مفید باشد.
کمک میکند Kafka را بهعنوان سیستم زنده و پیچیده تصور کنیم (بعد از همه distributed است) که ممکن است گاهی به مراقبت و feeding از وقت به وقت نیاز داشته باشد. در بخش بعد، testing topic را tackle میکنیم.
۷.۳ تست با EmbeddedKafkaCluster
با همهٔ گزینههای پیکربندی که داریم، خوب است آنها را هم تست کنیم. چه میشد اگر بتوانیم cluster Kafka را بدون داشتن cluster production-ready واقعی بالا بیاوریم؟ Kafka Streams کلاس utility integration به نام EmbeddedKafkaCluster را فراهم میکند که بین mock objectها و cluster کامل جای میگیرد. این کلاس cluster Kafka in-memory فراهم میکند [13]. اگرچه با Kafka Streams در ذهن ساخته شده، میتوانیم از آن برای تست clientهای Kafka استفاده کنیم.
لیستینگ ۷-۶ شبیه تستهای یافتشده در کتاب Kafka Streams in Action اثر William P. Bejeck Jr. تنظیم شده، مثلاً کلاس KafkaStreamsYellingIntegrationTest [14]. آن کتاب و کتاب بعدی او، Event Streaming with Kafka Streams and ksqlDB، مثالهای تست عمیقتر نشان میدهند. پیشنهاد میکنیم آنها را ببینید، از جمله پیشنهادش برای استفاده از Testcontainers (https://www.testcontainers.org/). لیستینگ بعدی تست با EmbeddedKafkaCluster و JUnit 4 را نشان میدهد.
لیستینگ ۷-۶. تست با EmbeddedKafkaCluster
java
@ClassRule
public static final EmbeddedKafkaCluster embeddedKafkaCluster
= new EmbeddedKafkaCluster(BROKER_NUMBER); // Uses JUnit-specific annotation
// to create the cluster with a
private Properties kaProducerProperties;
// specific number of brokers
private Properties kaConsumerProperties;
@Before
public void setUpBeforeClass() throws Exception {
embeddedKafkaCluster.createTopic(TOPIC,
PARTITION_NUMBER, REPLICATION_NUMBER);
kaProducerProperties = TestUtils.producerConfig(
embeddedKafkaCluster.bootstrapServers(),
AlertKeySerde.class,
StringSerializer.class);
kaConsumerProperties = TestUtils.consumerConfig( // Sets the consumer
embeddedKafkaCluster.bootstrapServers(), // configuration to point to the
AlertKeySerde.class, // embedded cluster brokers
StringDeserializer.class);
}
@Test
public void testAlertPartitioner() throws InterruptedException {
AlertProducer alertProducer = new AlertProducer();
try {
alertProducer.sendMessage(kaProducerProperties);
} catch (Exception ex) {
fail("kinaction_error EmbeddedKafkaCluster exception"
➥ + ex.getMessage()); // Calls the client without
} // any changes, which is
AlertConsumer alertConsumer = new AlertConsumer(); // clueless of the underlying
ConsumerRecords<Alert, String> records = // cluster being embedded
alertConsumer.getAlertMessages(kaConsumerProperties);
TopicPartition partition = new TopicPartition(TOPIC, 0);
List<ConsumerRecord<Alert, String>> results = records.records(partition);
assertEquals(0, results.get(0).partition()); // Asserts that the embedded
} // cluster handled the message
// from production to consumptionوقتی با EmbeddedKafkaCluster تست میکنیم، یکی از مهمترین بخشهای setup اطمینان از این است که cluster embedded قبل از شروع تست واقعی بالا آمده باشد. چون این cluster موقت است، نکتهٔ کلیدی دیگر اطمینان از این است که clientهای producer و consumer بدانند چگونه به این cluster in-memory اشاره کنند. برای کشف آن endpointها، میتوانیم از متد bootstrapServers() استفاده کنیم تا پیکربندی لازم را به clientها بدهیم. inject کردن آن پیکربندی به instanceهای client دوباره به استرategی پیکربندی شما بستگی دارد، اما میتواند بهسادگی تنظیم مقادیر با یک method call باشد. علاوه بر این پیکربندیها، clientها باید بتوانند بدون نیاز به فراهم کردن ویژگیهای mock Kafka تست کنند!
تست در لیستینگ ۷-۶ verify میکند منطق AlertLevelPartitioner درست بود. استفاده از آن منطق partitioner سفارشی با پیام critical باید alert را روی پارتیشن ۰ با کد مثال فصل ۴ مینشاند. با retrieve کردن پیامها برای TopicPartition(TOPIC, 0) و نگاه به پیامهای شاملشده، محل پارتیشن پیام تأیید شد. بهطور کلی، این سطح تست معمولاً integration testing در نظر گرفته میشود و شما را فراتر از فقط یک component تحت تست میبرد. در این نقطه، منطق client خودمان را با یک cluster Kafka تست کردهایم و بیش از یک module را integrate کردهایم.
یادداشت مطمئن شوید به تغییرات
pom.xmlدر source code برای فصل ۷ reference میکنید. JARهای مختلفی وجود دارند که در فصلهای قبلی لازم نبودند. همچنین، بعضی JARها فقط با classifierهای خاص include شدهاند، با یادآوری که فقط برای سناریوهای test لازماند.
۷.۳.۱ استفاده از Kafka Testcontainers
اگر متوجه شدید باید infrastructure را create و سپس tear down کنید، یکی از گزینههایی که میتوانید استفاده کنید (بهویژه برای integration testing) Testcontainers است (https://www.testcontainers.org/modules/kafka/). این کتابخانهٔ Java از Docker و یکی از frameworkهای تست JVM مختلف مثل JUnit استفاده میکند. Testcontainers به Docker imageها وابسته است تا cluster در حال اجرا به شما بدهد. اگر workflow شما Docker-based است یا تکنیک توسعهای که تیم شما خوب از آن استفاده میکند، Testcontainers ارزش دارد برای راهاندازی cluster Kafka برای تست بررسی شود.
یادداشت یکی از coauthorهای این کتاب، Viktor Gamov، repositoryی (https://github.com/gAmUssA/testcontainers-java-module-confluent-platform) از integration testing اجزای Confluent Platform (شامل Kafka، Schema Registry، ksqlDB) را نگه میدارد.
۷.۴ فشردهسازی topic
حالا که پایهٔ محکمی روی topicها که از پارتیشنها تشکیل شدهاند و پارتیشنها که از segmentها تشکیل شدهاند داریم، وقت صحبت دربارهٔ جزئیات log compaction است.
با compaction، هدف expire کردن پیامها نیست بلکه اطمینان از این است که latest value برای یک key وجود دارد و state قبلی نگه داشته نمیشود. همانطور که تازه reference شد، compaction به key بودن بخشی از پیامها و non-null بودن آن key وابسته است [10].
گزینهٔ پیکربندی که برای ایجاد topic فشردهشونده استفاده کردیم cleanup.policy=compact است [15]. این با مقدار پیکربندی پیشفرض که قبل از override روی delete تنظیم شده بود متفاوت است. بهعبارت دیگر، باید انتخاب کنیم topic فشردهشونده بسازیم وگرنه topic به آن شکل وجود نخواهد داشت. لیستینگ بعدی گزینهٔ پیکربندی لازم برای این topic فشردهشوندهٔ جدید را اضافه میکند.
لیستینگ ۷-۷. ایجاد topic فشردهشونده
bash
# Creates the topic like
# any other topic
bin/kafka-topics.sh --create --bootstrap-server localhost:9094 \
--topic kinaction_compact --partitions 3 --replication-factor 3 \
--config cleanup.policy=compact # Creates our topic type
# to be compactedیکی از سادهترین مقایسهها برای نحوهٔ ارائهٔ داده توسط topic فشردهشونده را میتوان در نحوهٔ بهروزرسانی فیلد موجود آرایه توسط کد دید، بهجای append کردن دادهٔ بیشتر. فرض کنید میخواهیم وضعیت عضویت فعلی برای یک عضویت آنلاین نگه داریم. کاربر فقط میتواند در یک state باشد، یا Basic یا Gold. در ابتدا کاربر در طرح Basic ثبتنام میکند، اما با گذشت زمان برای ویژگیهای بیشتر به طرح Gold ارتقا میدهد. اگرچه این هنوز رویدادی است که Kafka ذخیره میکند، در مورد ما فقط آخرین سطح عضویت برای مشتری خاص (key ما) مهم است. شکل ۷-۷ مثالی با استفاده از سه مشتری نشان میدهد.
شکل ۷-۷. compaction بهطور کلی
Log segment: Precompaction Compacted topic Offset Key Value Offset Key Value 0 Customer 0 Basic 2 Customer 0 Gold 1 Customer 1 Gold 3 Customer 2 Basic 2 Customer 0 Gold 100 Customer 1 Basic 3 Customer 2 Basic . . . . . . . . . 100 Customer 1 Basic
بعد از انجام compaction، آخرین بهروزرسانی customer 0 (در مثال ما) تنها چیزی است که در topic وجود دارد. پیامی با offset 2 مقدار قدیمی Basic (پیام offset 0) را برای customer 0 با Gold جایگزین میکند. Customer 1 مقدار فعلی Basic دارد چون latest key-specific offset برابر 100، state Gold قبلی offset 1 را بهروز میکند. چون customer 2 فقط یک رویداد دارد، آن رویداد بدون تغییر به topic فشردهشونده منتقل میشود.
مثال دنیای واقعی دیگر برای چرا کسی بخواهد از topic فشردهشونده استفاده کند، topic داخلی Kafka یعنی __consumer_offsets است. Kafka به history offsetهایی که یک consumer group مصرف میکند نیاز ندارد؛ فقط latest offset را میخواهد. با ذخیرهٔ offsetها در topic فشردهشونده، log در اصل view بهروزی از state فعلی دنیایش میگیرد.
وقتی topic برای compaction علامتگذاری میشود، میتوانیم یک log را در چند state مختلف ببینیم: فشردهشده یا نه. برای segmentهای قدیمیتر، duplicate valueها برای هر key باید بعد از تکمیل compaction به فقط یک value کاهش یافته باشند. پیامهای segment فعال آنهایی هستند که هنوز compaction نشدهاند [11]. چند value برای یک پیام با key مشخص میتواند وجود داشته باشد تا همهٔ پیامها پاک شوند. شکل ۷-۸ نشان میدهد pointer چگونه برای نشان دادن پیامهایی که با compaction پردازش شدهاند و پیامهایی که هنوز visit نشدهاند استفاده میشود [16].
شکل ۷-۸. پاکسازی compaction
Compacted topic Offset Key Value Offset 0-Customer 0 3 Customer 2 Basic . . . . . . The duplicate . . . keys have Offset 1-Customer 1 100 Customer 1 Basic been removed. 101 Customer 0 Null This compaction is similar to the kinaction_alert topic compaction we used with a different key. Already cleaned marker
با نگاه دقیق به offsetها در شکل ۷-۸، میبینیم شکاف در شماره offsetهای segment پاکشده وجود دارد. چون پیامهای duplicate key فقط با latest value باقی میمانند، ممکن است بعضی شماره offset از segment file حذف شوند؛ برای مثال، offset 2 حذف شد. در بخشهای active، احتمالاً offset numberهای ever-increasing که به آن عادت داریم را میبینیم، بدون پرش تصادفی.
حالا به subscriberای switch میکنیم که میخواست حسابش را حذف کند. با ارسال رویدادی با key مشترک، مثل Customer 0، با message value برابر null، این پیام بهعنوان delete treated میشود. این پیام tombstone در نظر گرفته میشود [10]. اگر از سیستمهای دیگر مثل Apache HBase™ استفاده کردهاید، مفهوم مشابه است. شکل ۷-۹ نشان میدهد value null پیام را حذف نمیکند بلکه مثل هر پیام دیگر serve میشود [10].
شکل ۷-۹. compaction برای value حذفشده
Delete with null Log segment: Precompaction Compacted topic Offset Key Value Offset Key Value 0 Customer 0 Basic 3 Customer 2 Basic . . . 1 Customer 1 Gold . . . . . . 2 Customer 0 Gold 100 Customer 1 Basic 3 Customer 2 Basic . . . 101 Customer 0 Null . . . . . . 100 Customer 1 Basic 101 Customer 0 Null Customer 0 isn't deleted immediately.
با قوانین delete که برنامه ممکن است یا نباید با آنها کنار بیاید، Kafka میتواند با feature set اصلی خود به برآورده کردن آن الزامات داده کمک کند.
در طول این فصل، جزئیات مختلف topicها، پارتیشنها و segmentها را بررسی کردیم. اگرچه broker-specific هستند، واقعاً میتوانند بر clientهای ما تأثیر بگذارند. چون اکنون تجربهای داریم از نحوهٔ ذخیرهٔ برخی دادههای خود Kafka، در فصل بعد وقت میگذاریم برای بحث دربارهٔ نحوهٔ ذخیرهٔ دادهٔ خودمان. این شامل گزینههای ذخیرهسازی بلندمدت برای داده است.
خلاصه
- Topicها غیرملموس هستند نه ساختار فیزیکی. برای فهم رفتار topic، consumer آن topic باید دربارهٔ تعداد پارتیشنها و replication factorهای در جریان بداند.
- پارتیشنها topicها را تشکیل میدهند و واحد پایه برای پردازش موازی داده داخل topic هستند.
- segmentهای فایل log در دایرکتوری پارتیشنها نوشته میشوند و توسط broker مدیریت میشوند.
- تست میتواند برای اعتبارسنجی منطق پارتیشن استفاده شود و ممکن است از cluster in-memory استفاده کند.
- فشردهسازی topic راهی برای ارائهٔ نمای latest value یک رکورد مشخص است.
منابع
- «Main Concepts and Terminology.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka/introduction.html#main-concepts-and-terminology (accessed August 28, 2021).
- J. Rao. «How to choose the number of topics/partitions in a Kafka cluster?» (March 12, 2015). Confluent blog. https://www.confluent.io/blog/how-choose-number-topics-partitions-kafka-cluster/ (accessed May 19, 2019).
- «Documentation: Modifying topics.» Apache Software Foundation (n.d.). https://kafka.apache.org/documentation/#basic_ops_modify_topic (accessed May 19, 2018).
- «Documentation: Adding and removing topics.» Apache Software Foundation (n.d.). https://kafka.apache.org/documentation/#basic_ops_add_topic (accessed December 11, 2019).
- «delete.topic.enable.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/broker-configs.html#brokerconfigs_delete.topic.enable (accessed January 15, 2021).
- Topics.java. Apache Kafka GitHub. https://github.com/apache/kafka/blob/99b9b3e84f4e98c3f07714e1de6a139a004cbc5b/clients/src/main/java/org/apache/kafka/common/internals/Topic.java (accessed August 27, 2021).
- «auto.create.topics.enable.» Apache Software Foundation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/broker-configs.html#brokerconfigs_auto.create.topics.enable (accessed December 19, 2019).
- AdminUtils.scala. Apache Kafka GitHub. https://github.com/apache/kafka/blob/d9b898b678158626bd2872bbfef883ca60a41c43/core/src/main/scala/kafka/admin/AdminUtils.scala (accessed August 27, 2021).
- «Documentation: index.interval.bytes.» Apache Kafka documentation. https://kafka.apache.org/documentation/#topicconfigs_index.interval.bytes (accessed August 27, 2021).
- «Log Compaction.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka/design.html#log-compaction (accessed August 20, 2021).
- «Configuring The Log Cleaner.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka/design.html#configuring-the-log-cleaner (accessed August 27, 2021).
- «CLI Tools for Confluent Platform.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/cli-reference.html (accessed August 25, 2021).
- EmbeddedKafkaCluster.java. Apache Kafka GitHub. https://github.com/apache/kafka/blob/9af81955c497b31b211b1e21d8323c875518df39/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java (accessed August 27, 2021).
- W. P. Bejeck Jr. Kafka Streams in Action. Shelter Island, NY, USA: Manning, 2018.
- «cleanup.policy.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/installation/configuration/topic-configs.html#topicconfigs_cleanup.policy (accessed November 22, 2020).
- «Log Compaction Basics.» Confluent documentation (n.d.). https://docs.confluent.io/platform/current/kafka/design.html#log-compaction-basics (accessed August 20, 2021).