കാഫ്കയിൽ നിന്ന് ലഭിച്ച ഇവന്റുകൾ റീപ്രോസസ് ചെയ്യുന്നു

ഹലോ, ഹബ്ർ.

അടുത്തിടെ ഐ തന്റെ അനുഭവം പങ്കുവെച്ചു കാഫ്ക പ്രൊഡ്യൂസറിനും ഉപഭോക്താവിനും ഗ്യാരണ്ടീഡ് ഡെലിവറിയുമായി അടുക്കാൻ ഞങ്ങൾ ഒരു ടീം എന്ന നിലയിൽ ഏതൊക്കെ പാരാമീറ്ററുകളാണ് മിക്കപ്പോഴും ഉപയോഗിക്കുന്നത് എന്നതിനെക്കുറിച്ച്. ബാഹ്യ സംവിധാനത്തിന്റെ താൽക്കാലിക ലഭ്യതക്കുറവിന്റെ ഫലമായി കാഫ്കയിൽ നിന്ന് ലഭിച്ച ഒരു ഇവന്റിന്റെ പുനഃസംസ്കരണം ഞങ്ങൾ എങ്ങനെയാണ് സംഘടിപ്പിച്ചതെന്ന് ഈ ലേഖനത്തിൽ ഞാൻ നിങ്ങളോട് പറയാൻ ആഗ്രഹിക്കുന്നു.

ആധുനിക ആപ്ലിക്കേഷനുകൾ വളരെ സങ്കീർണ്ണമായ അന്തരീക്ഷത്തിലാണ് പ്രവർത്തിക്കുന്നത്. ഒരു ആധുനിക ടെക്നോളജി സ്റ്റാക്കിൽ പൊതിഞ്ഞ ബിസിനസ്സ് ലോജിക്, Kubernetes അല്ലെങ്കിൽ OpenShift പോലുള്ള ഒരു ഓർക്കസ്ട്രേറ്റർ നിയന്ത്രിക്കുന്ന ഒരു ഡോക്കർ ഇമേജിൽ പ്രവർത്തിക്കുന്നു, കൂടാതെ ഫിസിക്കൽ, വെർച്വൽ റൂട്ടറുകളുടെ ഒരു ശൃംഖലയിലൂടെ മറ്റ് ആപ്ലിക്കേഷനുകളുമായോ എന്റർപ്രൈസ് സൊല്യൂഷനുകളുമായോ ആശയവിനിമയം നടത്തുന്നു. അത്തരമൊരു പരിതസ്ഥിതിയിൽ, എന്തെങ്കിലും എപ്പോഴും തകരാറിലായേക്കാം, അതിനാൽ ബാഹ്യ സംവിധാനങ്ങളിലൊന്ന് ലഭ്യമല്ലെങ്കിൽ ഇവന്റുകൾ വീണ്ടും പ്രോസസ്സ് ചെയ്യുന്നത് ഞങ്ങളുടെ ബിസിനസ്സ് പ്രക്രിയകളുടെ ഒരു പ്രധാന ഭാഗമാണ്.

കാഫ്കയ്ക്ക് മുമ്പ് എങ്ങനെയായിരുന്നു

നേരത്തെ പ്രോജക്റ്റിൽ ഞങ്ങൾ അസിൻക്രണസ് സന്ദേശ വിതരണത്തിനായി IBM MQ ഉപയോഗിച്ചിരുന്നു. സേവനത്തിന്റെ പ്രവർത്തന സമയത്ത് എന്തെങ്കിലും പിശക് സംഭവിച്ചാൽ, ലഭിച്ച സന്ദേശം കൂടുതൽ മാനുവൽ പാഴ്‌സിംഗിനായി ഒരു ഡെഡ്-ലെറ്റർ-ക്യൂവിൽ (DLQ) സ്ഥാപിക്കാം. ഇൻകമിംഗ് ക്യൂവിന് അടുത്തായി DLQ സൃഷ്ടിച്ചു, സന്ദേശം IBM MQ- ലേക്ക് കൈമാറി.

പിശക് താൽകാലികമാണെങ്കിൽ ഞങ്ങൾക്ക് അത് നിർണ്ണയിക്കാൻ കഴിയുമെങ്കിൽ (ഉദാഹരണത്തിന്, ഒരു HTTP കോളിലെ ഒരു ResourceAccessException അല്ലെങ്കിൽ MongoDb അഭ്യർത്ഥനയിൽ ഒരു MongoTimeoutException), പിന്നെ വീണ്ടും ശ്രമിക്കുന്നതിനുള്ള തന്ത്രം പ്രാബല്യത്തിൽ വരും. ആപ്ലിക്കേഷന്റെ ബ്രാഞ്ചിംഗ് ലോജിക് പരിഗണിക്കാതെ തന്നെ, യഥാർത്ഥ സന്ദേശം അയയ്‌ക്കാൻ കാലതാമസം നേരിടുന്ന സിസ്റ്റം ക്യൂവിലേക്കോ സന്ദേശങ്ങൾ വീണ്ടും അയയ്‌ക്കുന്നതിന് വളരെ മുമ്പുതന്നെ ഉണ്ടാക്കിയ ഒരു പ്രത്യേക അപ്ലിക്കേഷനിലേക്കോ മാറ്റി. ഇതിൽ സന്ദേശ തലക്കെട്ടിലെ ഒരു റീസെൻഡ് നമ്പർ ഉൾപ്പെടുന്നു, അത് കാലതാമസം ഇടവേളയുമായോ ആപ്ലിക്കേഷൻ-ലെവൽ സ്ട്രാറ്റജിയുടെ അവസാനവുമായോ ബന്ധപ്പെട്ടിരിക്കുന്നു. ഞങ്ങൾ തന്ത്രത്തിന്റെ അവസാനത്തിൽ എത്തിയിട്ടുണ്ടെങ്കിലും ബാഹ്യ സിസ്റ്റം ഇപ്പോഴും ലഭ്യമല്ലെങ്കിൽ, സ്വമേധയാ പാഴ്‌സിംഗിനായി സന്ദേശം DLQ-ൽ സ്ഥാപിക്കും.

ഒരു പരിഹാരം കണ്ടെത്തുന്നു

ഇന്റർനെറ്റിൽ തിരയുന്നു, നിങ്ങൾക്ക് ഇനിപ്പറയുന്നവ കണ്ടെത്താനാകും തീരുമാനം. ചുരുക്കത്തിൽ, ഓരോ കാലതാമസ ഇടവേളയ്‌ക്കും ഒരു വിഷയം സൃഷ്‌ടിക്കാനും ആവശ്യമായ കാലതാമസമുള്ള സന്ദേശങ്ങൾ വായിക്കുന്ന വശത്ത് ഉപഭോക്തൃ അപ്ലിക്കേഷനുകൾ നടപ്പിലാക്കാനും നിർദ്ദേശിക്കുന്നു.

ധാരാളം പോസിറ്റീവ് അവലോകനങ്ങൾ ഉണ്ടായിരുന്നിട്ടും, ഇത് പൂർണ്ണമായും വിജയിച്ചില്ലെന്ന് എനിക്ക് തോന്നുന്നു. ഒന്നാമതായി, കാരണം ഡവലപ്പർ, ബിസിനസ്സ് ആവശ്യകതകൾ നടപ്പിലാക്കുന്നതിനു പുറമേ, വിവരിച്ച സംവിധാനം നടപ്പിലാക്കാൻ ധാരാളം സമയം ചെലവഴിക്കേണ്ടിവരും.

കൂടാതെ, കാഫ്ക ക്ലസ്റ്ററിൽ ആക്സസ് കൺട്രോൾ പ്രവർത്തനക്ഷമമാക്കിയിട്ടുണ്ടെങ്കിൽ, വിഷയങ്ങൾ സൃഷ്ടിക്കാനും അവയിലേക്ക് ആവശ്യമായ ആക്സസ് നൽകാനും നിങ്ങൾ കുറച്ച് സമയം ചെലവഴിക്കേണ്ടിവരും. ഇതുകൂടാതെ, ഓരോ പുനരവലോകന വിഷയങ്ങൾക്കുമായി നിങ്ങൾ ശരിയായ retention.ms പാരാമീറ്റർ തിരഞ്ഞെടുക്കേണ്ടതുണ്ട്, അതുവഴി സന്ദേശങ്ങൾ വീണ്ടും അയയ്‌ക്കാനും അതിൽ നിന്ന് അപ്രത്യക്ഷമാകാതിരിക്കാനും സമയമുണ്ട്. നിലവിലുള്ള അല്ലെങ്കിൽ പുതിയ ഓരോ സേവനത്തിനും പ്രാവർത്തികമാക്കലും പ്രവേശന അഭ്യർത്ഥനയും ആവർത്തിക്കേണ്ടതുണ്ട്.

മെസേജ് റീപ്രോസസിങ്ങിനായി നമുക്ക് പൊതുവെ സ്പ്രിംഗ് സംവിധാനങ്ങളും പ്രത്യേകിച്ച് സ്പ്രിംഗ്-കാഫ്കയും എന്തൊക്കെയാണ് നൽകുന്നത് എന്ന് നോക്കാം. സ്പ്രിംഗ്-കാഫ്കയ്ക്ക് സ്പ്രിംഗ്-റിട്രിയിൽ ഒരു ട്രാൻസിറ്റീവ് ഡിപൻഡൻസി ഉണ്ട്, ഇത് വ്യത്യസ്ത ബാക്ക്ഓഫ് പോളിസികൾ കൈകാര്യം ചെയ്യുന്നതിനുള്ള സംഗ്രഹങ്ങൾ നൽകുന്നു. ഇത് തികച്ചും വഴക്കമുള്ള ഉപകരണമാണ്, എന്നാൽ അതിന്റെ പ്രധാന പോരായ്മ ആപ്ലിക്കേഷൻ മെമ്മറിയിൽ വീണ്ടും അയയ്‌ക്കുന്നതിന് സന്ദേശങ്ങൾ സംഭരിക്കുന്നു എന്നതാണ്. ഇതിനർത്ഥം, ഒരു അപ്‌ഡേറ്റ് അല്ലെങ്കിൽ പ്രവർത്തന പിശക് കാരണം ആപ്ലിക്കേഷൻ പുനരാരംഭിക്കുന്നത്, റീപ്രോസസ് ചെയ്യാൻ ശേഷിക്കുന്ന എല്ലാ സന്ദേശങ്ങളും നഷ്‌ടപ്പെടുന്നതിന് ഇടയാക്കും എന്നാണ്. ഈ പോയിന്റ് ഞങ്ങളുടെ സിസ്റ്റത്തിന് നിർണായകമായതിനാൽ, ഞങ്ങൾ അത് കൂടുതൽ പരിഗണിച്ചില്ല.

സ്പ്രിംഗ്-കാഫ്ക തന്നെ കണ്ടെയ്നർഅവെയർ എറർഹാൻഡ്‌ലറിന്റെ നിരവധി നിർവ്വഹണങ്ങൾ നൽകുന്നു, ഉദാഹരണത്തിന് SeekToCurrentErrorHandler, ഒരു പിശകുണ്ടായാൽ ഓഫ്‌സെറ്റ് മാറ്റാതെ തന്നെ നിങ്ങൾക്ക് സന്ദേശം പിന്നീട് പ്രോസസ്സ് ചെയ്യാൻ കഴിയും. സ്പ്രിംഗ്-കാഫ്ക 2.3 പതിപ്പിൽ തുടങ്ങി, BackOffPolicy സജ്ജീകരിക്കാൻ സാധിച്ചു.

ആപ്ലിക്കേഷൻ പുനരാരംഭിക്കുന്നതിനെ അതിജീവിക്കാൻ റീപ്രോസസ് ചെയ്ത സന്ദേശങ്ങളെ ഈ സമീപനം അനുവദിക്കുന്നു, പക്ഷേ ഇപ്പോഴും 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
    }

ഉദാഹരണത്തിൽ, MessageListenerContainer ഇപ്പോഴും പ്രവർത്തിക്കുന്നുണ്ടെങ്കിൽ അത് പുനരാരംഭിക്കാനുള്ള സമയമാണ് retryAt. TaskScheduler-ൽ സമാരംഭിച്ച ഒരു പ്രത്യേക ത്രെഡിൽ വീണ്ടും സമാരംഭം സംഭവിക്കും, ഇത് നടപ്പിലാക്കുന്നത് സ്പ്രിംഗ് വഴിയാണ്.

ഇനിപ്പറയുന്ന രീതിയിൽ ഞങ്ങൾ retryAt മൂല്യം കണ്ടെത്തുന്നു:

1. റീ-കോൾ കൗണ്ടറിന്റെ മൂല്യം നോക്കുന്നു.
2. കൌണ്ടർ മൂല്യത്തെ അടിസ്ഥാനമാക്കി, വീണ്ടും ശ്രമിക്കുന്നതിനുള്ള തന്ത്രത്തിലെ നിലവിലെ കാലതാമസം ഇടവേള തിരയുന്നു. തന്ത്രം ആപ്ലിക്കേഷനിൽ തന്നെ പ്രഖ്യാപിച്ചിരിക്കുന്നു; അത് സംഭരിക്കാൻ ഞങ്ങൾ JSON ഫോർമാറ്റ് തിരഞ്ഞെടുത്തു.
3. JSON അറേയിൽ കണ്ടെത്തിയ ഇടവേളയിൽ പ്രോസസ്സിംഗ് ആവർത്തിക്കേണ്ട സെക്കൻഡുകളുടെ എണ്ണം അടങ്ങിയിരിക്കുന്നു. retryAt-നുള്ള മൂല്യം രൂപപ്പെടുത്തുന്നതിന് ഈ സെക്കന്റുകളുടെ എണ്ണം നിലവിലെ സമയത്തിലേക്ക് ചേർത്തു.
4. ഇടവേള കണ്ടെത്തിയില്ലെങ്കിൽ, retryAt-ന്റെ മൂല്യം അസാധുവാണ്, സ്വമേധയാ പാഴ്‌സിംഗിനായി സന്ദേശം DLQ-ലേക്ക് അയയ്‌ക്കും.

ഈ സമീപനത്തിലൂടെ, നിലവിൽ പ്രോസസ്സ് ചെയ്യുന്ന ഓരോ സന്ദേശത്തിനും ആവർത്തിച്ചുള്ള കോളുകളുടെ എണ്ണം സംരക്ഷിക്കുക മാത്രമാണ് ശേഷിക്കുന്നത്, ഉദാഹരണത്തിന് ആപ്ലിക്കേഷൻ മെമ്മറിയിൽ. ഒരു ലീനിയർ ലോജിക് ആപ്ലിക്കേഷന് പ്രോസസ്സിംഗ് മൊത്തത്തിൽ കൈകാര്യം ചെയ്യാൻ കഴിയാത്തതിനാൽ, ഈ സമീപനത്തിന് റീട്രി കൗണ്ട് മെമ്മറിയിൽ സൂക്ഷിക്കുന്നത് നിർണായകമല്ല. സ്പ്രിംഗ്-റിട്രിയിൽ നിന്ന് വ്യത്യസ്തമായി, ആപ്ലിക്കേഷൻ പുനരാരംഭിക്കുന്നത് എല്ലാ സന്ദേശങ്ങളും നഷ്‌ടപ്പെടുന്നതിന് കാരണമാകില്ല, പക്ഷേ തന്ത്രം പുനരാരംഭിക്കും.

ഈ സമീപനം ബാഹ്യ സിസ്റ്റത്തിൽ നിന്ന് ലോഡ് എടുക്കാൻ സഹായിക്കുന്നു, വളരെ കനത്ത ലോഡ് കാരണം ഇത് ലഭ്യമല്ലായിരിക്കാം. മറ്റൊരു വിധത്തിൽ പറഞ്ഞാൽ, റീപ്രോസസ് ചെയ്യുന്നതിനു പുറമേ, ഞങ്ങൾ പാറ്റേൺ നടപ്പിലാക്കി സർക്യൂട്ട് ബ്രേക്കർ.

ഞങ്ങളുടെ കാര്യത്തിൽ, പിശക് പരിധി 1 മാത്രമാണ്, താൽക്കാലിക നെറ്റ്‌വർക്ക് തകരാറുകൾ കാരണം സിസ്റ്റം പ്രവർത്തനരഹിതമായ സമയം കുറയ്ക്കുന്നതിന്, ചെറിയ ലേറ്റൻസി ഇടവേളകളുള്ള വളരെ ഗ്രാനുലാർ റീട്രി സ്ട്രാറ്റജി ഞങ്ങൾ ഉപയോഗിക്കുന്നു. ഇത് എല്ലാ ഗ്രൂപ്പ് ആപ്ലിക്കേഷനുകൾക്കും അനുയോജ്യമാകണമെന്നില്ല, അതിനാൽ സിസ്റ്റത്തിന്റെ സവിശേഷതകളെ അടിസ്ഥാനമാക്കി പിശക് പരിധിയും ഇടവേള മൂല്യവും തമ്മിലുള്ള ബന്ധം തിരഞ്ഞെടുക്കണം.

നോൺ-ഡിറ്റർമിനിസ്റ്റിക് ലോജിക് ഉള്ള ആപ്ലിക്കേഷനുകളിൽ നിന്നുള്ള സന്ദേശങ്ങൾ പ്രോസസ്സ് ചെയ്യുന്നതിനുള്ള ഒരു പ്രത്യേക ആപ്ലിക്കേഷൻ

RETRY_AT സമയം എത്തുമ്പോൾ DESTINATION വിഷയത്തിലേക്ക് വീണ്ടും അയയ്‌ക്കുന്ന അത്തരം ഒരു അപ്ലിക്കേഷനിലേക്ക് (Retryer) ഒരു സന്ദേശം അയയ്‌ക്കുന്ന കോഡിന്റെ ഒരു ഉദാഹരണം ഇതാ:

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 മൂല്യം അപ്‌ഡേറ്റ് ചെയ്‌തു.
• RETRY_AT-ൽ എത്തുമ്പോൾ വീണ്ടും പ്രോസസ്സ് ചെയ്യുന്നതിന് സന്ദേശം അയച്ചതാണോ അതോ DLQ-ൽ സ്ഥാപിച്ചിട്ടുണ്ടോ എന്ന് സൂചിപ്പിക്കുന്ന സ്ഥിരാങ്കമാണ് SEND_TO.
• കാരണം - സന്ദേശ പ്രോസസ്സിംഗ് തടസ്സപ്പെട്ടതിന്റെ കാരണം.

PostgreSQL-ൽ വീണ്ടും അയയ്‌ക്കുന്നതിനും സ്വമേധയാ പാഴ്‌സിംഗ് ചെയ്യുന്നതിനുമുള്ള സന്ദേശങ്ങൾ റീട്രയർ സംഭരിക്കുന്നു. ഒരു ടൈമർ, RETRY_AT ഉപയോഗിച്ച് സന്ദേശങ്ങൾ കണ്ടെത്തുന്ന ഒരു ടാസ്‌ക് ആരംഭിക്കുകയും കീ record.key() ഉപയോഗിച്ച് DESTINATION വിഷയത്തിന്റെ ORIGINAL_PARTITION പാർട്ടീഷനിലേക്ക് തിരികെ അയയ്ക്കുകയും ചെയ്യുന്നു.

അയച്ചുകഴിഞ്ഞാൽ, PostgreSQL-ൽ നിന്ന് സന്ദേശങ്ങൾ ഇല്ലാതാക്കപ്പെടും. REST API വഴി Retryer-മായി സംവദിക്കുന്ന ഒരു ലളിതമായ UI-ലാണ് സന്ദേശങ്ങളുടെ സ്വമേധയാ പാഴ്‌സിംഗ് നടക്കുന്നത്. DLQ-ൽ നിന്ന് സന്ദേശങ്ങൾ വീണ്ടും അയയ്‌ക്കുകയോ ഇല്ലാതാക്കുകയോ ചെയ്യുക, പിശക് വിവരങ്ങൾ കാണുക, സന്ദേശങ്ങൾക്കായി തിരയുക എന്നിവയാണ് ഇതിന്റെ പ്രധാന സവിശേഷതകൾ, ഉദാഹരണത്തിന് പിശക് പേര്.

ഞങ്ങളുടെ ക്ലസ്റ്ററുകളിൽ ആക്‌സസ് കൺട്രോൾ പ്രവർത്തനക്ഷമമാക്കിയിരിക്കുന്നതിനാൽ, റീട്രയർ കേൾക്കുന്ന വിഷയത്തിലേക്ക് അധികമായി ആക്‌സസ് അഭ്യർത്ഥിക്കുകയും DESTINATION വിഷയത്തിലേക്ക് എഴുതാൻ റീട്രയറെ അനുവദിക്കുകയും ചെയ്യേണ്ടത് ആവശ്യമാണ്. ഇത് അസൗകര്യമാണ്, പക്ഷേ, ഇന്റർവെൽ ടോപ്പിക്ക് സമീപനത്തിൽ നിന്ന് വ്യത്യസ്തമായി, ഇത് മാനേജ് ചെയ്യാൻ ഞങ്ങൾക്ക് ഒരു പൂർണ്ണമായ DLQ, UI ഉണ്ട്.

ഒരു ഇൻകമിംഗ് വിഷയം നിരവധി വ്യത്യസ്ത ഉപഭോക്തൃ ഗ്രൂപ്പുകൾ വായിക്കുന്ന സന്ദർഭങ്ങളുണ്ട്, അവരുടെ ആപ്ലിക്കേഷനുകൾ വ്യത്യസ്ത യുക്തികൾ നടപ്പിലാക്കുന്നു. ഈ ആപ്ലിക്കേഷനുകളിലൊന്നിന് റീട്രയർ വഴി ഒരു സന്ദേശം വീണ്ടും പ്രോസസ്സ് ചെയ്യുന്നത് മറ്റൊന്നിന്റെ ഡ്യൂപ്ലിക്കേറ്റിന് കാരണമാകും. ഇതിൽ നിന്ന് പരിരക്ഷിക്കുന്നതിന്, വീണ്ടും പ്രോസസ്സിംഗിനായി ഞങ്ങൾ ഒരു പ്രത്യേക വിഷയം സൃഷ്ടിക്കുന്നു. ഇൻകമിംഗ്, റീട്രൈ വിഷയങ്ങൾ ഒരേ ഉപഭോക്താവിന് യാതൊരു നിയന്ത്രണവുമില്ലാതെ വായിക്കാൻ കഴിയും.

ഡിഫോൾട്ടായി ഈ സമീപനം സർക്യൂട്ട് ബ്രേക്കർ പ്രവർത്തനക്ഷമത നൽകുന്നില്ല, എന്നിരുന്നാലും ഇത് ഉപയോഗിച്ച് ആപ്ലിക്കേഷനിൽ ചേർക്കാവുന്നതാണ് സ്പ്രിംഗ്-ക്ലൗഡ്-നെറ്റ്ഫ്ലിക്സ് അല്ലെങ്കിൽ പുതിയത് സ്പ്രിംഗ് ക്ലൗഡ് സർക്യൂട്ട് ബ്രേക്കർ, ബാഹ്യ സേവനങ്ങൾ വിളിക്കപ്പെടുന്ന സ്ഥലങ്ങളെ ഉചിതമായ സംഗ്രഹങ്ങളിലേക്ക് പൊതിയുന്നു. കൂടാതെ, ഒരു തന്ത്രം തിരഞ്ഞെടുക്കുന്നത് സാധ്യമാകും ബൾക്ക്ഹെഡ് പാറ്റേൺ, അത് ഉപയോഗപ്രദമാകും. ഉദാഹരണത്തിന്, സ്പ്രിംഗ്-ക്ലൗഡ്-നെറ്റ്ഫ്ലിക്സിൽ ഇത് ഒരു ത്രെഡ് പൂൾ അല്ലെങ്കിൽ സെമാഫോർ ആകാം.

തീരുമാനം

തൽഫലമായി, ഏതെങ്കിലും ബാഹ്യ സിസ്റ്റം താൽക്കാലികമായി ലഭ്യമല്ലെങ്കിൽ സന്ദേശ പ്രോസസ്സിംഗ് ആവർത്തിക്കാൻ ഞങ്ങളെ അനുവദിക്കുന്ന ഒരു പ്രത്യേക ആപ്ലിക്കേഷനുണ്ട്.

ഒരേ കാഫ്ക ക്ലസ്റ്ററിൽ പ്രവർത്തിക്കുന്ന ബാഹ്യ സംവിധാനങ്ങൾക്ക് അവയുടെ ഭാഗത്ത് കാര്യമായ മാറ്റങ്ങളില്ലാതെ ഇത് ഉപയോഗിക്കാൻ കഴിയും എന്നതാണ് ആപ്ലിക്കേഷന്റെ പ്രധാന നേട്ടങ്ങളിലൊന്ന്! അത്തരമൊരു ആപ്ലിക്കേഷന് വീണ്ടും ശ്രമിക്കാനുള്ള വിഷയം ആക്സസ് ചെയ്യാനും കുറച്ച് കാഫ്ക തലക്കെട്ടുകൾ പൂരിപ്പിച്ച് റീട്രയറിന് ഒരു സന്ദേശം അയയ്ക്കാനും മാത്രമേ ആവശ്യമുള്ളൂ. അധിക അടിസ്ഥാന സൗകര്യങ്ങളൊന്നും ഉയർത്തേണ്ട ആവശ്യമില്ല. ആപ്ലിക്കേഷനിൽ നിന്ന് റീട്രയറിലേക്കും തിരിച്ചും കൈമാറ്റം ചെയ്യപ്പെടുന്ന സന്ദേശങ്ങളുടെ എണ്ണം കുറയ്ക്കുന്നതിന്, ഞങ്ങൾ ലീനിയർ ലോജിക്കിലുള്ള ആപ്ലിക്കേഷനുകൾ തിരിച്ചറിയുകയും അവ കൺസ്യൂമർ സ്റ്റോപ്പിലൂടെ വീണ്ടും പ്രോസസ്സ് ചെയ്യുകയും ചെയ്തു.

അവലംബം: www.habr.com