કાફકા તરફથી મળેલી ઘટનાઓની પુનઃપ્રક્રિયા

હે હબર.

તાજેતરમાં આઇ પોતાનો અનુભવ શેર કર્યો બાંયધરીકૃત ડિલિવરીની નજીક જવા માટે અમે કાફકા નિર્માતા અને ઉપભોક્તા માટે મોટાભાગે ટીમ તરીકે કયા પરિમાણોનો ઉપયોગ કરીએ છીએ તે વિશે. આ લેખમાં હું તમને જણાવવા માંગુ છું કે અમે બાહ્ય સિસ્ટમની અસ્થાયી અનુપલબ્ધતાના પરિણામે કાફકા તરફથી પ્રાપ્ત ઇવેન્ટની પુનઃપ્રક્રિયા કેવી રીતે ગોઠવી.

આધુનિક એપ્લિકેશનો ખૂબ જટિલ વાતાવરણમાં કાર્ય કરે છે. આધુનિક ટેક્નોલોજી સ્ટેકમાં લપેટાયેલ વ્યવસાય તર્ક, કુબરનેટ્સ અથવા ઓપનશિફ્ટ જેવા ઓર્કેસ્ટ્રેટર દ્વારા સંચાલિત ડોકર ઇમેજમાં ચાલે છે અને ભૌતિક અને વર્ચ્યુઅલ રાઉટર્સની સાંકળ દ્વારા અન્ય એપ્લિકેશનો અથવા એન્ટરપ્રાઇઝ સોલ્યુશન્સ સાથે વાતચીત કરે છે. આવા વાતાવરણમાં, કંઈક હંમેશા તૂટી શકે છે, તેથી જો કોઈ બાહ્ય સિસ્ટમ ઉપલબ્ધ ન હોય તો ઘટનાઓને પુનઃપ્રક્રિયા કરવી એ અમારી વ્યવસાય પ્રક્રિયાઓનો એક મહત્વપૂર્ણ ભાગ છે.

કાફકા પહેલા કેવું હતું

અગાઉ પ્રોજેક્ટમાં અમે અસિંક્રોનસ મેસેજ ડિલિવરી માટે 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 એ MessageListenerContainer ને પુનઃપ્રારંભ કરવાનો સમય છે જો તે હજુ પણ ચાલુ છે. ટાસ્કશેડ્યુલરમાં લૉન્ચ કરાયેલા એક અલગ થ્રેડમાં ફરીથી લૉન્ચ થશે, જેનું અમલીકરણ પણ વસંત દ્વારા પ્રદાન કરવામાં આવે છે.

અમે નીચેની રીતે retryAt મૂલ્ય શોધીએ છીએ:

1. રી-કોલ કાઉન્ટરની કિંમત ઉપર જોવામાં આવે છે.
2. કાઉન્ટર વેલ્યુના આધારે, પુનઃપ્રયાસ વ્યૂહરચનામાં વર્તમાન વિલંબ અંતરાલ શોધવામાં આવે છે. વ્યૂહરચના એપ્લિકેશનમાં જ જાહેર કરવામાં આવી છે; અમે તેને સંગ્રહિત કરવા માટે JSON ફોર્મેટ પસંદ કર્યું છે.
3. JSON એરેમાં મળેલ અંતરાલમાં સેકંડની સંખ્યા છે જેના પછી પ્રક્રિયાને પુનરાવર્તિત કરવાની જરૂર પડશે. retryAt માટે મૂલ્ય બનાવવા માટે આ સેકન્ડની સંખ્યા વર્તમાન સમયમાં ઉમેરવામાં આવે છે.
4. જો અંતરાલ ન મળે, તો 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 માંથી સંદેશાઓને ફરીથી મોકલવા અથવા કાઢી નાખવા, ભૂલની માહિતી જોવી અને સંદેશાઓ શોધવાનું છે, ઉદાહરણ તરીકે ભૂલના નામ દ્વારા.

અમારા ક્લસ્ટરો પર એક્સેસ કંટ્રોલ સક્ષમ હોવાથી, પુન: પ્રયાસ કરનાર જે વિષયને સાંભળી રહ્યો છે તેની ઍક્સેસની વિનંતી કરવી અને પુનઃપ્રયત્નકર્તાને DESTINATION વિષય પર લખવાની મંજૂરી આપવી જરૂરી છે. આ અસુવિધાજનક છે, પરંતુ, ઇન્ટરવલ વિષયના અભિગમથી વિપરીત, અમારી પાસે તેનું સંચાલન કરવા માટે સંપૂર્ણ સુવિધાયુક્ત DLQ અને UI છે.

એવા કિસ્સાઓ છે જ્યારે ઇનકમિંગ વિષય ઘણા જુદા જુદા ઉપભોક્તા જૂથો દ્વારા વાંચવામાં આવે છે, જેમની એપ્લિકેશનો વિવિધ તર્કનો અમલ કરે છે. આમાંની એક એપ્લિકેશન માટે Retryer દ્વારા સંદેશને પુનઃપ્રક્રિયા કરવાથી બીજી ડુપ્લિકેટમાં પરિણમશે. આની સામે રક્ષણ આપવા માટે, અમે ફરીથી પ્રક્રિયા કરવા માટે એક અલગ વિષય બનાવીએ છીએ. આવનારા અને પુનઃપ્રયાસના વિષયો સમાન ઉપભોક્તા કોઈપણ પ્રતિબંધ વિના વાંચી શકે છે.

મૂળભૂત રીતે આ અભિગમ સર્કિટ બ્રેકર કાર્યક્ષમતા પ્રદાન કરતું નથી, જો કે તેનો ઉપયોગ કરીને એપ્લિકેશનમાં ઉમેરી શકાય છે spring-Cloud-netflix અથવા નવું વસંત વાદળ સર્કિટ બ્રેકર, સ્થાનોને લપેટીને જ્યાં બાહ્ય સેવાઓને યોગ્ય અમૂર્તમાં બોલાવવામાં આવે છે. વધુમાં, તે માટે વ્યૂહરચના પસંદ કરવાનું શક્ય બને છે બલ્કહેડ પેટર્ન, જે ઉપયોગી પણ હોઈ શકે છે. ઉદાહરણ તરીકે, સ્પ્રિંગ-ક્લાઉડ-નેટફ્લિક્સમાં આ થ્રેડ પૂલ અથવા સેમાફોર હોઈ શકે છે.

નિષ્કર્ષ

પરિણામે, અમારી પાસે એક અલગ એપ્લિકેશન છે જે અમને સંદેશ પ્રક્રિયાને પુનરાવર્તિત કરવાની મંજૂરી આપે છે જો કોઈપણ બાહ્ય સિસ્ટમ અસ્થાયી રૂપે અનુપલબ્ધ હોય.

એપ્લિકેશનનો એક મુખ્ય ફાયદો એ છે કે તેનો ઉપયોગ સમાન કાફકા ક્લસ્ટર પર ચાલતી બાહ્ય સિસ્ટમો દ્વારા કરી શકાય છે, તેમની બાજુમાં નોંધપાત્ર ફેરફારો કર્યા વિના! આવી એપ્લિકેશનને ફક્ત પુનઃપ્રયાસના વિષયને ઍક્સેસ કરવાની જરૂર પડશે, થોડા કાફકા હેડરો ભરો અને ફરીથી પ્રયાસ કરનારને સંદેશ મોકલો. કોઈ વધારાનું ઈન્ફ્રાસ્ટ્રક્ચર ઊભું કરવાની જરૂર નથી. અને એપ્લીકેશનમાંથી રીટ્રીયર અને બેક પર ટ્રાન્સફર કરાયેલા સંદેશાઓની સંખ્યા ઘટાડવા માટે, અમે રેખીય તર્ક સાથે એપ્લિકેશનોને ઓળખી અને ઉપભોક્તા સ્ટોપ દ્વારા ફરીથી પ્રક્રિયા કરી.

સોર્સ: www.habr.com