🥇 কাফকা থেকে প্রাপ্ত ইভেন্টগুলি পুনরায় প্রক্রিয়াকরণ

আরে হাবর।

সম্প্রতি আমি তার অভিজ্ঞতা শেয়ার করেছেন গ্যারান্টিযুক্ত ডেলিভারির কাছাকাছি যেতে আমরা কাফকা প্রযোজক এবং ভোক্তাদের জন্য দল হিসাবে কোন প্যারামিটারগুলি প্রায়শই ব্যবহার করি। এই নিবন্ধে আমি আপনাকে বলতে চাই কিভাবে আমরা বহিরাগত সিস্টেমের অস্থায়ী অনুপলব্ধতার ফলে কাফকা থেকে প্রাপ্ত একটি ইভেন্টের পুনরায় প্রক্রিয়াকরণের আয়োজন করেছি।

আধুনিক অ্যাপ্লিকেশনগুলি খুব জটিল পরিবেশে কাজ করে। একটি আধুনিক প্রযুক্তির স্ট্যাকে মোড়ানো ব্যবসায়িক যুক্তি, কুবারনেটস বা ওপেনশিফটের মতো একজন অর্কেস্ট্রেটর দ্বারা পরিচালিত একটি ডকার ইমেজে চলছে এবং শারীরিক এবং ভার্চুয়াল রাউটারগুলির একটি চেইনের মাধ্যমে অন্যান্য অ্যাপ্লিকেশন বা এন্টারপ্রাইজ সমাধানগুলির সাথে যোগাযোগ করা। এই ধরনের পরিবেশে, কিছু সবসময় ভেঙে যেতে পারে, তাই বহিরাগত সিস্টেমগুলির মধ্যে একটি অনুপলব্ধ হলে ইভেন্টগুলি পুনঃপ্রসেস করা আমাদের ব্যবসায়িক প্রক্রিয়াগুলির একটি গুরুত্বপূর্ণ অংশ।

কাফকার আগে কেমন ছিল

এর আগে প্রজেক্টে আমরা অ্যাসিঙ্ক্রোনাস মেসেজ ডেলিভারির জন্য IBM MQ ব্যবহার করতাম। পরিষেবার অপারেশন চলাকালীন কোনো ত্রুটি দেখা দিলে, প্রাপ্ত বার্তাটিকে আরও ম্যানুয়াল পার্সিংয়ের জন্য একটি ডেড-লেটার-কিউ (DLQ) এ স্থাপন করা যেতে পারে। DLQ আগত সারির পাশে তৈরি করা হয়েছিল, বার্তাটি IBM MQ এর ভিতরে স্থানান্তরিত হয়েছিল।

যদি ত্রুটিটি অস্থায়ী হয় এবং আমরা এটি নির্ধারণ করতে পারি (উদাহরণস্বরূপ, একটি HTTP কলে একটি ResourceAccessException বা MongoDb অনুরোধে একটি MongoTimeoutException), তাহলে পুনরায় চেষ্টা করার কৌশল কার্যকর হবে৷ অ্যাপ্লিকেশনটির শাখাগত যুক্তি নির্বিশেষে, মূল বার্তাটি বিলম্বিত পাঠানোর জন্য সিস্টেম সারিতে স্থানান্তরিত হয়েছিল, অথবা একটি পৃথক অ্যাপ্লিকেশনে যা বার্তাগুলি পুনরায় পাঠানোর জন্য অনেক আগে তৈরি করা হয়েছিল৷ এতে বার্তা শিরোনামে একটি পুনরায় পাঠানো নম্বর অন্তর্ভুক্ত রয়েছে, যা বিলম্বের ব্যবধান বা অ্যাপ্লিকেশন-স্তরের কৌশলের শেষের সাথে আবদ্ধ। যদি আমরা কৌশলের শেষ প্রান্তে পৌঁছে যাই কিন্তু বাহ্যিক সিস্টেম এখনও অনুপলব্ধ হয়, তাহলে ম্যানুয়াল পার্সিংয়ের জন্য বার্তাটি DLQ-তে স্থাপন করা হবে।

সমাধান অনুসন্ধান করুন

ইন্টারনেটে অনুসন্ধান করা হচ্ছে, আপনি নিম্নলিখিত খুঁজে পেতে পারেন সিদ্ধান্ত. সংক্ষেপে, প্রতিটি বিলম্বের ব্যবধানের জন্য একটি বিষয় তৈরি করার এবং পাশে ভোক্তা অ্যাপ্লিকেশনগুলি বাস্তবায়নের প্রস্তাব করা হয়েছে, যা প্রয়োজনীয় বিলম্বের সাথে বার্তাগুলি পড়বে।

বিপুল সংখ্যক ইতিবাচক পর্যালোচনা সত্ত্বেও, এটি আমার কাছে সম্পূর্ণরূপে সফল নয় বলে মনে হয়। প্রথমত, কারণ বিকাশকারীকে, ব্যবসায়ের প্রয়োজনীয়তা বাস্তবায়নের পাশাপাশি, বর্ণিত প্রক্রিয়াটি বাস্তবায়নে অনেক সময় ব্যয় করতে হবে।

উপরন্তু, যদি কাফকা ক্লাস্টারে অ্যাক্সেস কন্ট্রোল সক্ষম করা থাকে, তাহলে আপনাকে বিষয়গুলি তৈরি করতে এবং সেগুলিতে প্রয়োজনীয় অ্যাক্সেস প্রদান করতে কিছু সময় ব্যয় করতে হবে। এটি ছাড়াও, আপনাকে প্রতিটি পুনঃপ্রচারের বিষয়গুলির জন্য সঠিক retention.ms প্যারামিটার নির্বাচন করতে হবে যাতে বার্তাগুলি পুনরায় পাঠানোর সময় থাকে এবং এটি থেকে অদৃশ্য না হয়৷ প্রতিটি বিদ্যমান বা নতুন পরিষেবার জন্য প্রয়োগ এবং অ্যাক্সেসের অনুরোধ পুনরাবৃত্তি করতে হবে।

এখন দেখা যাক সাধারণভাবে স্প্রিং এবং স্প্রিং-কাফকা বিশেষ করে বার্তা পুনঃপ্রক্রিয়াকরণের জন্য আমাদের কী ব্যবস্থা প্রদান করে। স্প্রিং-কাফকার স্প্রিং-রিট্রির উপর একটি ট্রানজিটিভ নির্ভরতা রয়েছে, যা বিভিন্ন ব্যাকঅফ পলিসি পরিচালনার জন্য বিমূর্ততা প্রদান করে। এটি একটি মোটামুটি নমনীয় টুল, কিন্তু এর উল্লেখযোগ্য ত্রুটি হল অ্যাপ্লিকেশন মেমরিতে পুনরায় পাঠানোর জন্য বার্তা সংরক্ষণ করা। এর মানে হল যে একটি আপডেট বা একটি অপারেশনাল ত্রুটির কারণে অ্যাপ্লিকেশনটি পুনরায় চালু করার ফলে পুনঃপ্রসেসিং মুলতুবি থাকা সমস্ত বার্তা হারিয়ে যাবে৷ যেহেতু এই পয়েন্টটি আমাদের সিস্টেমের জন্য গুরুত্বপূর্ণ, আমরা এটিকে আর বিবেচনা করিনি।

spring-kafka নিজেই ContainerAwareErrorHandler-এর বেশ কয়েকটি বাস্তবায়ন প্রদান করে, উদাহরণস্বরূপ SeekToCurrentErrorHandler, যা দিয়ে আপনি একটি ত্রুটির ক্ষেত্রে অফসেট স্থানান্তর না করে পরে বার্তাটি প্রক্রিয়া করতে পারেন। স্প্রিং-কাফকা 2.3 সংস্করণ দিয়ে শুরু করে, ব্যাকঅফপলিসি সেট করা সম্ভব হয়েছিল।

এই পদ্ধতিটি পুনঃপ্রক্রিয়াজাত বার্তাগুলিকে অ্যাপ্লিকেশন পুনঃসূচনা থেকে বাঁচতে দেয়, তবে এখনও কোনও DLQ প্রক্রিয়া নেই। আমরা 2019 এর শুরুতে এই বিকল্পটি বেছে নিয়েছিলাম, আশাবাদীভাবে বিশ্বাস করে যে DLQ এর প্রয়োজন হবে না (আমরা ভাগ্যবান ছিলাম এবং এই ধরনের রিপ্রসেসিং সিস্টেমের সাথে অ্যাপ্লিকেশনটি পরিচালনা করার কয়েক মাস পরে আসলে এটির প্রয়োজন ছিল না)। সাময়িক ত্রুটির কারণে SeekToCurrentErrorHandler ফায়ার হয়েছে৷ অবশিষ্ট ত্রুটিগুলি লগে প্রিন্ট করা হয়েছে, যার ফলে একটি অফসেট হয়েছে, এবং পরবর্তী বার্তার সাথে প্রক্রিয়াকরণ অব্যাহত রয়েছে।

চূড়ান্ত সিদ্ধান্ত

SeekToCurrentErrorHandler-এর উপর ভিত্তি করে বাস্তবায়ন আমাদের বার্তা পুনরায় পাঠানোর জন্য আমাদের নিজস্ব প্রক্রিয়া বিকাশ করতে প্ররোচিত করেছে।

প্রথমত, আমরা বিদ্যমান অভিজ্ঞতা ব্যবহার করতে এবং অ্যাপ্লিকেশন যুক্তির উপর নির্ভর করে এটি প্রসারিত করতে চেয়েছিলাম। একটি রৈখিক যুক্তি প্রয়োগের জন্য, পুনরায় চেষ্টা করার কৌশল দ্বারা নির্দিষ্ট সময়ের জন্য নতুন বার্তা পড়া বন্ধ করা সর্বোত্তম হবে। অন্যান্য অ্যাপ্লিকেশনের জন্য, আমি একটি একক পয়েন্ট রাখতে চেয়েছিলাম যা পুনরায় চেষ্টা করার কৌশল প্রয়োগ করবে। উপরন্তু, এই একক পয়েন্ট উভয় পদ্ধতির জন্য DLQ কার্যকারিতা থাকতে হবে।

পুনরায় চেষ্টা করার কৌশলটি অবশ্যই অ্যাপ্লিকেশনটিতে সংরক্ষণ করতে হবে, যা একটি অস্থায়ী ত্রুটি ঘটলে পরবর্তী বিরতি পুনরুদ্ধার করার জন্য দায়ী।

একটি লিনিয়ার লজিক অ্যাপ্লিকেশনের জন্য ভোক্তাকে থামানো

স্প্রিং-কাফকার সাথে কাজ করার সময়, ভোক্তাকে থামানোর কোডটি এইরকম দেখতে পারে:

public void pauseListenerContainer(MessageListenerContainer listenerContainer, 
                                   Instant retryAt) {
        if (nonNull(retryAt) && listenerContainer.isRunning()) {
            listenerContainer.stop();
            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
            return;
        }
        // to DLQ
    }

উদাহরণে, মেসেজলিস্টেনার কনটেইনারটি এখনও চালু থাকলে পুনরায় চালু করার সময় হল retryAt। টাস্কশিডুলারে চালু করা একটি পৃথক থ্রেডে পুনরায় লঞ্চ ঘটবে, যার বাস্তবায়নও বসন্ত দ্বারা সরবরাহ করা হয়।

আমরা নিম্নলিখিত উপায়ে retryAt মান খুঁজে পাই:

রি-কল কাউন্টারের মান দেখা হয়।
কাউন্টার মানের উপর ভিত্তি করে, পুনরায় চেষ্টা করার কৌশলে বর্তমান বিলম্বের ব্যবধান অনুসন্ধান করা হয়। কৌশলটি অ্যাপ্লিকেশনেই ঘোষণা করা হয়েছে; আমরা এটি সংরক্ষণ করার জন্য JSON বিন্যাসটি বেছে নিয়েছি।
JSON অ্যারেতে পাওয়া ব্যবধানে সেকেন্ডের সংখ্যা রয়েছে যার পরে প্রক্রিয়াকরণের পুনরাবৃত্তি করতে হবে। retryAt-এর মান তৈরি করতে বর্তমান সময়ের সাথে এই সেকেন্ডের সংখ্যা যোগ করা হয়।
যদি ব্যবধান পাওয়া না যায়, তাহলে retryAt-এর মান শূন্য হয় এবং ম্যানুয়াল পার্সিংয়ের জন্য বার্তাটি DLQ-এ পাঠানো হবে।

এই পদ্ধতির সাথে, যা অবশিষ্ট থাকে তা হল প্রতিটি বার্তার জন্য পুনরাবৃত্তি কলের সংখ্যা সংরক্ষণ করা যা বর্তমানে প্রক্রিয়া করা হচ্ছে, উদাহরণস্বরূপ অ্যাপ্লিকেশন মেমরিতে। মেমরিতে পুনরায় চেষ্টা গণনা রাখা এই পদ্ধতির জন্য গুরুত্বপূর্ণ নয়, যেহেতু একটি লিনিয়ার লজিক অ্যাপ্লিকেশন সম্পূর্ণরূপে প্রক্রিয়াকরণ পরিচালনা করতে পারে না। বসন্ত-পুনরায় চেষ্টার বিপরীতে, অ্যাপ্লিকেশনটি পুনরায় চালু করার ফলে সমস্ত বার্তাগুলি পুনরায় প্রক্রিয়াকরণের জন্য হারিয়ে যাবে না, তবে কেবল কৌশলটি পুনরায় চালু হবে।

এই পদ্ধতিটি বাহ্যিক সিস্টেমের লোডটি বন্ধ করতে সহায়তা করে, যা খুব ভারী লোডের কারণে অনুপলব্ধ হতে পারে। অন্য কথায়, পুনঃপ্রক্রিয়াকরণ ছাড়াও, আমরা প্যাটার্নের বাস্তবায়ন অর্জন করেছি সার্কিট ব্রেকার.

আমাদের ক্ষেত্রে, ত্রুটির থ্রেশহোল্ড মাত্র 1, এবং অস্থায়ী নেটওয়ার্ক বিভ্রাটের কারণে সিস্টেম ডাউনটাইম কমানোর জন্য, আমরা ছোট লেটেন্সি ব্যবধান সহ একটি খুব দানাদার পুনঃচেষ্টা কৌশল ব্যবহার করি। এটি সমস্ত গ্রুপ অ্যাপ্লিকেশনের জন্য উপযুক্ত নাও হতে পারে, তাই সিস্টেমের বৈশিষ্ট্যের উপর ভিত্তি করে ত্রুটি থ্রেশহোল্ড এবং ব্যবধানের মানের মধ্যে সম্পর্ক নির্বাচন করা আবশ্যক।

নন-ডিটারমিনিস্টিক যুক্তি সহ অ্যাপ্লিকেশন থেকে বার্তা প্রক্রিয়াকরণের জন্য একটি পৃথক অ্যাপ্লিকেশন

এখানে কোডের একটি উদাহরণ রয়েছে যা এই ধরনের একটি অ্যাপ্লিকেশনে একটি বার্তা পাঠায় (পুনরায় চেষ্টাকারী), যা RETRY_AT সময় পৌঁছে গেলে DESTINATION বিষয়ে পুনরায় পাঠানো হবে:


public <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, 
                         Instant retryAt, String counter, String groupId, Exception e) {
        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
        List<Header> arrayOfHeaders = 
            new ArrayList<>(Arrays.asList(headers.toArray()));
        updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
        updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
        updateHeader(arrayOfHeaders, ORIGINAL_PARTITION, 
                     () -> Integer.toString(record.partition()).getBytes());
        if (nonNull(retryAt)) {
            updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
            updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
        } else {
            updateHeader(arrayOfHeaders, REASON, 
                         ExceptionUtils.getStackTrace(e)::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
        }
        ProducerRecord<K, V> messageToSend =
            new ProducerRecord<>(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
        kafkaTemplate.send(messageToSend);
    }

উদাহরণ দেখায় যে শিরোনামে প্রচুর তথ্য প্রেরণ করা হয়। RETRY_AT-এর মান একইভাবে পাওয়া যায় যেমনটি কনজিউমার স্টপের মাধ্যমে পুনরায় চেষ্টা করার পদ্ধতির জন্য। DESTINATION এবং RETRY_AT ছাড়াও আমরা পাস করি:

GROUP_ID, যার দ্বারা আমরা ম্যানুয়াল বিশ্লেষণ এবং সরলীকৃত অনুসন্ধানের জন্য বার্তাগুলিকে গোষ্ঠীভুক্ত করি৷
ORIGINAL_PARTITION পুনরায় প্রক্রিয়াকরণের জন্য একই ভোক্তা রাখার চেষ্টা করতে। এই প্যারামিটারটি শূন্য হতে পারে, এই ক্ষেত্রে মূল বার্তার record.key() কী ব্যবহার করে নতুন পার্টিশন পাওয়া যাবে।
পুনরায় চেষ্টা করার কৌশল অনুসরণ করতে COUNTER মান আপডেট করা হয়েছে।
SEND_TO হল একটি ধ্রুবক যা নির্দেশ করে যে বার্তাটি RETRY_AT-এ পৌঁছানোর পরে পুনরায় প্রক্রিয়াকরণের জন্য পাঠানো হয়েছে বা DLQ-এ রাখা হয়েছে।
REASON - যে কারণে বার্তা প্রক্রিয়াকরণ বাধাগ্রস্ত হয়েছিল।

পুনঃপ্রচারকারী পোস্টগ্রেএসকিউএল-এ পুনরায় পাঠানো এবং ম্যানুয়াল পার্সিংয়ের জন্য বার্তা সংরক্ষণ করে। একটি টাইমার একটি কাজ শুরু করে যা RETRY_AT-এর মাধ্যমে বার্তাগুলি খুঁজে পায় এবং কী record.key() দিয়ে DESTINATION বিষয়ের ORIGINAL_PARTITION পার্টিশনে ফেরত পাঠায়।

একবার পাঠানো হলে, PostgreSQL থেকে বার্তা মুছে ফেলা হয়। বার্তাগুলির ম্যানুয়াল পার্সিং একটি সাধারণ UI এ ঘটে যা REST API এর মাধ্যমে Retryer এর সাথে ইন্টারঅ্যাক্ট করে। এর প্রধান বৈশিষ্ট্যগুলি হল DLQ থেকে বার্তাগুলি পুনরায় পাঠানো বা মুছে ফেলা, ত্রুটির তথ্য দেখা এবং বার্তাগুলি অনুসন্ধান করা, উদাহরণস্বরূপ ত্রুটির নাম দ্বারা।

যেহেতু আমাদের ক্লাস্টারে অ্যাক্সেস কন্ট্রোল সক্ষম করা আছে, তাই রিট্রায়ার যে বিষয়ে শুনছে সেই বিষয়ে অতিরিক্ত অ্যাক্সেসের অনুরোধ করা এবং রিট্রায়ারকে গন্তব্য বিষয়ে লিখতে অনুমতি দেওয়া প্রয়োজন। এটি অসুবিধাজনক, কিন্তু, ব্যবধান বিষয় পদ্ধতির বিপরীতে, এটি পরিচালনা করার জন্য আমাদের কাছে একটি পূর্ণাঙ্গ DLQ এবং UI আছে।

এমন কিছু ক্ষেত্রে রয়েছে যখন একটি আগত বিষয় বিভিন্ন ভোক্তা গোষ্ঠীর দ্বারা পড়া হয়, যাদের অ্যাপ্লিকেশনগুলি বিভিন্ন যুক্তি প্রয়োগ করে। এই অ্যাপ্লিকেশনগুলির একটির জন্য Retryer-এর মাধ্যমে একটি বার্তা পুনরায় প্রক্রিয়া করার ফলে অন্যটি একটি নকল হবে৷ এটি থেকে রক্ষা করার জন্য, আমরা পুনরায় প্রক্রিয়াকরণের জন্য একটি পৃথক বিষয় তৈরি করি। ইনকামিং এবং পুনরায় চেষ্টা করার বিষয়গুলি একই ভোক্তা কোন সীমাবদ্ধতা ছাড়াই পড়তে পারেন।

ডিফল্টরূপে এই পদ্ধতিটি সার্কিট ব্রেকার কার্যকারিতা প্রদান করে না, তবে এটি ব্যবহার করে অ্যাপ্লিকেশনটিতে যোগ করা যেতে পারে স্প্রিং-ক্লাউড-নেটফ্লিক্স বা নতুন স্প্রিং ক্লাউড সার্কিট ব্রেকার, সেই জায়গাগুলিকে মোড়ানো যেখানে বহিরাগত পরিষেবাগুলিকে উপযুক্ত বিমূর্ততায় ডাকা হয়৷ উপরন্তু, এটা জন্য একটি কৌশল চয়ন করা সম্ভব হয় বাল্কহেড প্যাটার্ন, যা দরকারী হতে পারে। উদাহরণস্বরূপ, স্প্রিং-ক্লাউড-নেটফ্লিক্সে এটি একটি থ্রেড পুল বা সেমাফোর হতে পারে।

উপসংহার

ফলস্বরূপ, আমাদের কাছে একটি পৃথক অ্যাপ্লিকেশন রয়েছে যা কোনো বহিরাগত সিস্টেম সাময়িকভাবে অনুপলব্ধ হলে আমাদের বার্তা প্রক্রিয়াকরণের পুনরাবৃত্তি করতে দেয়।

অ্যাপ্লিকেশনটির একটি প্রধান সুবিধা হল যে এটি একই কাফকা ক্লাস্টারে চলমান বহিরাগত সিস্টেমগুলির দ্বারা ব্যবহার করা যেতে পারে, তাদের পাশে উল্লেখযোগ্য পরিবর্তন ছাড়াই! এই ধরনের একটি অ্যাপ্লিকেশনের জন্য শুধুমাত্র পুনরায় চেষ্টা করার বিষয় অ্যাক্সেস করতে হবে, কয়েকটি কাফকা শিরোনাম পূরণ করতে হবে এবং পুনরায় চেষ্টাকারীকে একটি বার্তা পাঠাতে হবে। অতিরিক্ত কোনো অবকাঠামো বাড়াতে হবে না। এবং অ্যাপ্লিকেশন থেকে Retryer এবং পিছনে স্থানান্তরিত বার্তার সংখ্যা কমানোর জন্য, আমরা রৈখিক যুক্তি সহ অ্যাপ্লিকেশনগুলি চিহ্নিত করেছি এবং সেগুলিকে কনজিউমার স্টপের মাধ্যমে পুনরায় প্রক্রিয়া করেছি৷

উত্স: www.habr.com

কাফকার কাছ থেকে প্রাপ্ত ঘটনা পুনঃপ্রক্রিয়াকরণ