كافكا المرنة: ضغط الرجوع وطرود الرسائل الميتة

مقدمة

في البنى المعمارية الحديثة القائمة على الأحداث، يعمل Apache Kafka كنظام عصبي مركزي لتوزيع البيانات. ومع ذلك، فإن طبيعة "أرسل وانسَ" في تمرير الرسائل غالبًا ما تخفي الهشاشة الكامنة. عندما لا يتمكن المستهلكون من مواكبة المنتجين، أو عندما تفشل المعالجة بسبب أخطاء برمجية أو أعطال عابرة، قد تنهار الأنظمة. يستكشف هذا المنشور نمطين حاسمين للمرونة: إدارة ضغط الرجوع لمنع استنفاد الذاكرة، وتنفيذ طرود الرسائل الميتة (DLQs) لعزل الرسائل السامة التي تفشل في المعالجة.

فهم ضغط الرجوع في كافكا

ضغط الرجوع هو الآلية التي يُبلغ بها المستهلك الوسيط (Broker) بأنه مثقل. في كافكا، يُدار هذا بشكل رئيسي من خلال التزامات الإزاحات (Offset Commits) وأحجام الاسترداد (Fetch Sizes). إذا كانت المعالجة أبطأ من وصول الرسائل، فسيواصل الوسيط تخزين الرسائل في الموضوع، مما يوفر مخزنًا مؤقتًا بشكل فعال. ومع ذلك، يجب على التطبيق إدارة ذاكرته الداخلية. المفتاح للتعامل مع ضغط الرجوع لا يتعلق فقط بوسطاء كافكا، بل بمنطق المستهلك الخاص بك. إذا كان مستهلكك يقوم بعمليات إدخال/إخراج كثيفة (مثل الكتابة إلى قاعدة بيانات)، فسيتم حصره. بدون حدود مناسبة، يؤدي هذا إلى أخطاء OutOfMemoryError.

// مثال: تكوين حدود الاسترداد لإدارة الذاكرة
Properties props = new Properties();
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 5 * 1024 * 1024); // 5MB
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30000);

من خلال تحديد max.poll.records، تضمن أن دورة الاستطلاع الواحدة لا تسحب كمية كبيرة جدًا من البيانات إلى الذاكرة. إذا استغرقت المعالجة وقتًا أطول من max.poll.interval.ms، سيتم سحب المستهلك من المجموعة، مما يؤدي إلى إعادة توازن (Rebalance). هذا صمام أمان، لكن إعادة التوازن المتكررة مكلفة. الهدف هو ضبط هذه المعلمات بحيث يتمكن المستهلك من مواكبة معدل الاستيعاب دون إجهاده.

دور طرود الرسائل الميتة

ماذا يحدث عندما تفشل رسالة في المعالجة؟ استثناء بسيط مثل throw new Exception() في حلقة المستهلك سيتوقف المستهلك، مما يتسبب في تعطل الإزاحات. هذا يحجب القسم (Partition) بأكمله. لمنع هذا، نستخدم طرود الرسائل الميتة (DLQs). طرد الرسائل الميتة هو موضوع (أو طرد) منفصل حيث تُكتب الرسائل الفاشلة. يستمر المستهلك الأصلي في معالجة الرسالة التالية، بينما يحقق مستهلك طرد الرسائل الميتة (أو المشغل اليدوي) في سبب الفشل.

try {
    processMessage(record);
    // الالتزام بالإزاحة يدويًا أو الاعتماد على الالتزام التلقائي إذا نجحت المعالجة
} catch (Exception e) {
    // تسجيل الخطأ
    logger.error("Failed to process record: {}", record.value(), e);
    
    // الإرسال إلى طرد الرسائل الميتة
    dlqProducer.send(dlqTopic, record);
    
    // الالتزام بالإزاحة للسجل الأصلي للمتابعة
    consumer.commitSync(Map.of(topicPartition, new OffsetAndMetadata(record.offset() + 1)));
}

ملاحظة: في الإنتاج، يجب استخدام إطار عمل قوي مثل Spring Kafka أو وظيفة طرد الرسائل الميتة في Confluent Cloud للتعامل مع هذا المنطق المعاملاتي بشكل صحيح، مما يضمن أن الرسالة إما تُعالج أو تُنقل إلى طرد الرسائل الميتة دون تكرار أو فقدان.

استراتيجية التنفيذ العملية

1. مراقبة التأخر (Lag): تتبع تأخر مجموعة المستهلكين للكشف المبكر عن ضغط الرجوع. إذا زاد التأخر باستمرار، فهذا يشير إلى مشكلة في السعة. 2. القدرة على التكرار (Idempotency): تأكد من أن منطق المعالجة الخاص بك قابل للتكرار. نظرًا لأن عمليات الإعادة وإعادة إرسال طرد الرسائل الميتة يمكن أن تسبب تكرارًا، يجب أن يتعامل تطبيقك مع ذلك بسلاسة. 3. أوضاع التدهور (Degraded Modes): إذا كان الخدمة المنخفضة متوقفة، قرر ما إذا كنت ستسقط الرسائل (نادراً)، أو تخزنها محليًا (محفوف بالمخاطر)، أو توجهها إلى طرد الرسائل الميتة لإعادة تشغيلها لاحقًا.

خاتمة

بناء أنظمة مرنة قائمة على الأحداث يتطلب أكثر من مجرد نشر كافكا. إنه يتطلب ضبطًا دقيقًا لسلوك المستهلكين لإدارة ضغط الرجوع واستراتيجية قوية للتعامل مع الأعطال من خلال طرود الرسائل الميتة. من خلال اعتبار الرسائل بيانات هشة يمكن أن تفشل في أي مرحلة، تبني خطوطًا ليست قابلة للتوسع فحسب، بل قابلة للاسترداد أيضًا. ابدأ بمراقبة تأخرك وتنفيذ طرد رسائل ميتة أساسي لمواضيعك الأكثر أهمية.