تاپیک __consumer_offsets: مدیریت آفست‌های مصرف‌کنندگان

می دانیم که به ازای هر گروه مصرف کننده باید اطلاعات اینکه این گروه، از هر پارتیشن تا کدام آفست را مصرف کرده است ذخیره کنیم که اگر به هر دلیلی گروه ری‌استارت شد و یا عضو جدیدی به آن اضافه شد، کافکا دقیقا بداند از کدام آفست باید به این گروه یا مصرف کننده جدید، سرویس‌دهی کند.

قبلاً Kafka از ZooKeeper برای ذخیره مقادیر آفست‌های کامیت‌شده استفاده می‌کرد، به ازای هر Consumer Group و هر پارتیشن. اما از نسخه ۲.۸ به بعد، با غیر فعال کردن ZooKeeper، مدیریت ذخیره آفست‌ها توسط خود Kafka انجام می‌شود. این موضوع باعث شد که Kafka بتواند پیام‌ها و وضعیت مصرف‌کننده‌ها را به‌صورت داخلی و مطمئن مدیریت کند.

در کارگاه عملی، ما دو Consumer Group داشتیم و مشاهده کردیم که داده‌های آن‌ها در پارتیشن‌های ۱۵ و ۳۹ از تاپیک __consumer_offsets ذخیره می‌شود:

Partition=39 Offset=4732 Key=metadata_groupusers Value=...
Reached end of topic __consumer_offsets [15] at offset 4516
Reached end of topic __consumer_offsets [39] at offset 4733
Partition=15 Offset=4516 Key=metadata_group2users Value=...
Partition=15 Offset=4517 Key=metadata_group2users Value=...
...

نحوه ذخیره آفست‌ها

Kafka آفست مصرف‌کننده‌ها را در یک تاپیک داخلی و کامپکت‌شده به نام __consumer_offsets نگهداری می‌کند.

مشاهده تاپیک‌ها (شامل داخلی)

kafka-topics.sh --bootstrap-server localhost:9092 --list --include-internal

خروجی نمونه:

__consumer_offsets
users
users-stage-2
...

مشاهده پیام‌های آفست

پیام‌های تاپیک __consumer_offsets به صورت کامپکت‌شده و باینری هستند:

kcat -b localhost:9092 -t __consumer_offsets -C -o beginning -c 10 -f 'Partition=%p Offset=%o Key=%k Value=%s\n'

• Key شامل گروه، تاپیک، پارتیشن است
• Value شامل آخرین آفست کامیت‌شده و متادیتا می‌باشد
• Kafka با استفاده از این اطلاعات Consumer Group را دقیقاً از همان جایی که متوقف شده بود، ادامه می‌دهد

بررسی با kafka-consumer-groups.sh

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group metadata_group

• مقادیر کامیت شده فعلی با پیام‌های موجود در __consumer_offsets تطابق دارند

ارتباط با auto.offset.reset

تنظیم auto.offset.reset مشخص می‌کند که اگر هیچ آفست معتبری برای گروه مصرف‌کننده وجود نداشته باشد، مصرف‌کننده از کجا شروع به خواندن کند.

• earliest: مصرف از اولین پیام موجود در پارتیشن آغاز می‌شود، حتی اگر قبلاً آفستی ثبت نشده باشد.
• latest: مصرف‌کننده تنها پیام‌های جدیدی را که بعد از راه‌اندازی تولید می‌شوند، می‌خواند.
• none: اگر هیچ آفستی یافت نشود، خطا رخ می‌دهد.

⚠️ نکات مهم:

• earliest فقط زمانی اعمال می‌شود که هیچ آفست معتبری موجود نباشد. اگر آفست قبلاً ثبت شده باشد، Kafka از همان آفست شروع به خواندن می‌کند.
• استفاده از earliest می‌تواند باعث خواندن مجدد داده‌های قدیمی شود، که در سناریوهای بازسازی وضعیت مفید است ولی در سیستم‌های حجیم ممکن است بار اضافی ایجاد کند.
• باید توجه داشته باشید که سیاست‌های Retention می‌توانند روی داده‌های قدیمی تاثیر بگذارند و مصرف‌کننده‌ها ممکن است نتوانند از ابتدای پارتیشن بخوانند اگر داده‌ها حذف شده باشند.

نکته عملی

در کارگاه عملی مشاهده شد که دو Consumer Group ما داده‌های خود را در پارتیشن‌های ۱۵ و ۳۹ ذخیره کردند:

Partition=15 Offset=4516 Key=metadata_group2users Value=...
Partition=39 Offset=4733 Key=metadata_groupusers Value=...

• این نشان می‌دهد که Kafka داده های آفست های کامیت شده هر گروه را در یک پارتیشن از تاپیک `__consumer_offsets `ذخیره می‌کند.
• هنگام راه‌اندازی مجدد گروه‌ها، مصرف‌کننده‌ها دقیقاً از آفست ثبت شده ادامه می‌دهند.

✅ جمع‌بندی

• Kafka از تاپیک داخلی __consumer_offsets برای نگهداری آفست‌ها استفاده می‌کند.
• مصرف‌کننده‌ها با این تاپیک قادر به ادامه دقیق خواندن از همان جایی که متوقف شده‌اند هستند.
• تنظیم auto.offset.reset=earliest تعیین می‌کند که در صورت عدم وجود آفست معتبر، مصرف از اولین پیام موجود آغاز شود.
• سیاست‌های retention و مدیریت پارتیشن‌ها می‌تواند روی دسترسی به پیام‌های قدیمی تاثیر بگذارد.