കാഫ്ക എങ്ങനെ യാഥാർത്ഥ്യമായി

ഹേ ഹബർ!

സ്വന്തം അറിയിപ്പ് കേന്ദ്രം വികസിപ്പിക്കുന്ന ടിങ്കോഫ് ടീമിൽ ഞാൻ പ്രവർത്തിക്കുന്നു. ഞാൻ കൂടുതലും ജാവയിൽ സ്പ്രിംഗ് ബൂട്ട് ഉപയോഗിച്ച് വികസിപ്പിക്കുകയും ഒരു പ്രോജക്റ്റിൽ ഉണ്ടാകുന്ന വിവിധ സാങ്കേതിക പ്രശ്നങ്ങൾ പരിഹരിക്കുകയും ചെയ്യുന്നു.

ഞങ്ങളുടെ മിക്ക മൈക്രോ സർവീസുകളും ഒരു സന്ദേശ ബ്രോക്കർ വഴി പരസ്പരം അസമന്വിതമായി ആശയവിനിമയം നടത്തുന്നു. മുമ്പ്, ഞങ്ങൾ ഒരു ബ്രോക്കറായി IBM MQ ഉപയോഗിച്ചിരുന്നു, അത് മേലിൽ ലോഡിനെ നേരിടാൻ കഴിയില്ല, എന്നാൽ അതേ സമയം ഉയർന്ന ഡെലിവറി ഗ്യാരന്റി ഉണ്ടായിരുന്നു.

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

ഉറപ്പുള്ള ഡെലിവറിയും മറ്റും

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

ഇത് സഹായിക്കും client.id നിർമ്മാതാവിനും ഉപഭോക്താവിനും. ഒറ്റനോട്ടത്തിൽ, നിങ്ങൾക്ക് ആപ്ലിക്കേഷന്റെ പേര് മൂല്യമായി ഉപയോഗിക്കാം, മിക്ക കേസുകളിലും ഇത് പ്രവർത്തിക്കും. ഒരു ആപ്ലിക്കേഷൻ നിരവധി ഉപഭോക്താക്കൾ ഉപയോഗിക്കുകയും നിങ്ങൾ അവർക്ക് ഒരേ client.id നൽകുകയും ചെയ്യുന്ന സാഹചര്യമാണെങ്കിലും, ഇനിപ്പറയുന്ന മുന്നറിയിപ്പ് നൽകുന്നു:

org.apache.kafka.common.utils.AppInfoParser — Error registering AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0

കാഫ്കയുമായുള്ള ഒരു ആപ്ലിക്കേഷനിൽ നിങ്ങൾക്ക് JMX ഉപയോഗിക്കാൻ താൽപ്പര്യമുണ്ടെങ്കിൽ, ഇത് ഒരു പ്രശ്നമായേക്കാം. ഈ സാഹചര്യത്തിൽ, ആപ്ലിക്കേഷന്റെ പേരിന്റെ സംയോജനവും, ഉദാഹരണത്തിന്, വിഷയത്തിന്റെ പേര് client.id മൂല്യവും ഉപയോഗിക്കുന്നതാണ് നല്ലത്. ഞങ്ങളുടെ കോൺഫിഗറേഷന്റെ ഫലം കമാൻഡ് ഔട്ട്പുട്ടിൽ കാണാം കാഫ്ക-ഉപഭോക്തൃ ഗ്രൂപ്പുകൾ Confluent-ൽ നിന്നുള്ള യൂട്ടിലിറ്റികളിൽ നിന്ന്:

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

• 0 - അംഗീകാരം പരിഗണിക്കില്ല.
• 1 ആണ് ഡിഫോൾട്ട് പാരാമീറ്റർ, അംഗീകരിക്കാൻ 1 പകർപ്പ് മാത്രമേ ആവശ്യമുള്ളൂ.
• −1 — എല്ലാ സിൻക്രൊണൈസ് ചെയ്ത പകർപ്പുകളിൽ നിന്നുമുള്ള അംഗീകാരം ആവശ്യമാണ് (ക്ലസ്റ്റർ സജ്ജീകരണം min.insync.replicas).

ലിസ്റ്റുചെയ്ത മൂല്യങ്ങളിൽ നിന്ന്, −1 ന് തുല്യമായ ആക്കുകൾ സന്ദേശം നഷ്‌ടപ്പെടില്ല എന്നതിന് ശക്തമായ ഉറപ്പ് നൽകുന്നു.

നമുക്കെല്ലാവർക്കും അറിയാവുന്നതുപോലെ, വിതരണ സംവിധാനങ്ങൾ വിശ്വസനീയമല്ല. ക്ഷണികമായ പിഴവുകളിൽ നിന്ന് പരിരക്ഷിക്കുന്നതിന്, കാഫ്ക പ്രൊഡ്യൂസർ ഓപ്ഷൻ നൽകുന്നു വീണ്ടും ശ്രമിക്കുന്നു, ഉള്ളിൽ വീണ്ടും അയയ്ക്കാനുള്ള ശ്രമങ്ങളുടെ എണ്ണം സജ്ജമാക്കാൻ ഇത് നിങ്ങളെ അനുവദിക്കുന്നു delivery.timeout.ms. റീട്രീസ് പാരാമീറ്ററിന് Integer.MAX_VALUE (2147483647) ന്റെ ഡിഫോൾട്ട് മൂല്യമുള്ളതിനാൽ, delivery.timeout.ms മാത്രം മാറ്റിക്കൊണ്ട് സന്ദേശ പുനഃശ്രമങ്ങളുടെ എണ്ണം ക്രമീകരിക്കാൻ കഴിയും.

ഞങ്ങൾ കൃത്യമായി ഒരിക്കൽ ഡെലിവറിയിലേക്ക് നീങ്ങുകയാണ്

ലിസ്‌റ്റ് ചെയ്‌ത ക്രമീകരണങ്ങൾ ഉയർന്ന ഗ്യാരണ്ടിയോടെ സന്ദേശങ്ങൾ കൈമാറാൻ ഞങ്ങളുടെ നിർമ്മാതാവിനെ അനുവദിക്കുന്നു. കാഫ്ക വിഷയത്തിൽ ഒരു സന്ദേശത്തിന്റെ ഒരു പകർപ്പ് മാത്രമേ എഴുതിയിട്ടുള്ളൂ എന്ന് എങ്ങനെ ഉറപ്പാക്കാം എന്നതിനെക്കുറിച്ച് നമുക്ക് ഇപ്പോൾ സംസാരിക്കാം? ഏറ്റവും ലളിതമായ സാഹചര്യത്തിൽ, ഇത് ചെയ്യുന്നതിന്, നിങ്ങൾ പ്രൊഡ്യൂസറിൽ പാരാമീറ്റർ സജ്ജീകരിക്കേണ്ടതുണ്ട് പ്രവർത്തനക്ഷമമാക്കുക സത്യത്തിലേക്ക്. ഒരു വിഷയത്തിന്റെ ഒരു പ്രത്യേക പാർട്ടീഷനിൽ ഒരു സന്ദേശം മാത്രമേ എഴുതിയിട്ടുള്ളൂ എന്ന് ഐഡമ്പറ്റൻസി ഉറപ്പ് നൽകുന്നു. ഐഡമ്പറ്റൻസി പ്രാപ്തമാക്കുന്നതിനുള്ള മുൻവ്യവസ്ഥ മൂല്യങ്ങളാണ് acks = എല്ലാം, വീണ്ടും ശ്രമിക്കുക > 0, max.in.flight.requests.per.connection ≤ 5. ഈ പാരാമീറ്ററുകൾ ഡവലപ്പർ വ്യക്തമാക്കിയിട്ടില്ലെങ്കിൽ, മുകളിലുള്ള മൂല്യങ്ങൾ സ്വയമേവ സജ്ജീകരിക്കും.

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

• സന്ദേശം അയയ്ക്കുമ്പോൾ പാർട്ടീഷൻ വ്യക്തമായി വ്യക്തമാക്കിയിട്ടുണ്ടെങ്കിൽ, ഞങ്ങൾ അത് ഉപയോഗിക്കുന്നു.
• പാർട്ടീഷൻ വ്യക്തമാക്കിയിട്ടില്ലെങ്കിലും കീ വ്യക്തമാക്കിയിട്ടുണ്ടെങ്കിൽ, കീയുടെ ഹാഷ് ഉപയോഗിച്ച് പാർട്ടീഷൻ തിരഞ്ഞെടുക്കുക.
• പാർട്ടീഷനും കീയും വ്യക്തമാക്കിയിട്ടില്ലെങ്കിൽ, പാർട്ടീഷനുകൾ ഓരോന്നായി തിരഞ്ഞെടുക്കുക (റൗണ്ട്-റോബിൻ).

കൂടാതെ, ഒരു കീ ഉപയോഗിച്ച്, ഒരു പാരാമീറ്റർ ഉപയോഗിച്ച് അയയ്‌ക്കൽ max.in.flight.requests.per.connection = 1 ഉപഭോക്താവിൽ നിങ്ങൾക്ക് കാര്യക്ഷമമായ സന്ദേശ പ്രോസസ്സിംഗ് നൽകുന്നു. നിങ്ങളുടെ ക്ലസ്റ്ററിൽ ആക്‌സസ് കൺട്രോൾ കോൺഫിഗർ ചെയ്‌തിട്ടുണ്ടെങ്കിൽ, ഒരു വിഷയത്തിലേക്ക് അസാമാന്യമായി എഴുതാൻ നിങ്ങൾക്ക് അവകാശങ്ങൾ ആവശ്യമാണെന്നതും ഓർമിക്കേണ്ടതാണ്.

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

ഔപചാരികമായി, ആപ്ലിക്കേഷന്റെ പേര് പോലുള്ള ഏത് സ്ട്രിംഗും ഒരു ഇടപാട് ഐഡന്റിഫയറായി ഉപയോഗിക്കാം. എന്നാൽ നിങ്ങൾ ഒരേ ട്രാൻസാക്ഷൻ.ഐഡി ഉപയോഗിച്ച് ഒരേ ആപ്ലിക്കേഷന്റെ നിരവധി സന്ദർഭങ്ങൾ സമാരംഭിക്കുകയാണെങ്കിൽ, ആദ്യം സമാരംഭിച്ച സന്ദർഭം ഒരു പിശകോടെ നിർത്തും, കാരണം കാഫ്ക അതിനെ ഒരു സോംബി പ്രക്രിയയായി കണക്കാക്കും.

org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.

ഈ പ്രശ്നം പരിഹരിക്കാൻ, ഞങ്ങൾ ഹോസ്‌റ്റ് നെയിമിന്റെ രൂപത്തിൽ ആപ്ലിക്കേഷന്റെ പേരിലേക്ക് ഒരു പ്രത്യയം ചേർക്കുന്നു, അത് പരിസ്ഥിതി വേരിയബിളുകളിൽ നിന്ന് നമുക്ക് ലഭിക്കും.

പ്രൊഡ്യൂസർ കോൺഫിഗർ ചെയ്‌തിരിക്കുന്നു, എന്നാൽ കാഫ്കയിലെ ഇടപാടുകൾ സന്ദേശത്തിന്റെ വ്യാപ്തിയെ മാത്രമേ നിയന്ത്രിക്കൂ. ഇടപാട് നില പരിഗണിക്കാതെ തന്നെ, സന്ദേശം ഉടൻ തന്നെ വിഷയത്തിലേക്ക് പോകുന്നു, പക്ഷേ അധിക സിസ്റ്റം ആട്രിബ്യൂട്ടുകൾ ഉണ്ട്.

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

എന്നാൽ ഒരു ന്യൂനൻസ് കൂടിയുണ്ട്. ഞങ്ങൾ മുകളിൽ കോൺഫിഗർ ചെയ്‌ത Transactional.id യഥാർത്ഥത്തിൽ ഇടപാട് പ്രിഫിക്‌സാണ്. ഇടപാട് മാനേജറിൽ, ഒരു സീക്വൻസ് നമ്പർ അതിൽ ചേർത്തിരിക്കുന്നു. സ്വീകരിച്ച ഐഡന്റിഫയർ നൽകിയിരിക്കുന്നു Transactional.id.expiration.ms, ഒരു കാഫ്ക ക്ലസ്റ്ററിൽ കോൺഫിഗർ ചെയ്‌തിരിക്കുന്നതും "7 ദിവസം" എന്ന സ്ഥിര മൂല്യമുള്ളതുമാണ്. ഈ സമയത്ത് അപ്ലിക്കേഷന് സന്ദേശങ്ങളൊന്നും ലഭിച്ചിട്ടില്ലെങ്കിൽ, നിങ്ങൾ അടുത്ത ട്രാൻസാക്ഷനൽ അയയ്‌ക്കാൻ ശ്രമിക്കുമ്പോൾ നിങ്ങൾക്ക് ലഭിക്കും അസാധുവായ പിഡ്മാപ്പിംഗ് ഒഴിവാക്കൽ. ഇടപാട് കോർഡിനേറ്റർ അടുത്ത ഇടപാടിനായി ഒരു പുതിയ സീക്വൻസ് നമ്പർ നൽകും. എന്നിരുന്നാലും, InvalidPidMappingException ശരിയായി കൈകാര്യം ചെയ്തില്ലെങ്കിൽ സന്ദേശം നഷ്‌ടമായേക്കാം.

ഫലങ്ങൾക്ക് പകരം

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

നിർമ്മാണം:

1. acks = എല്ലാം
2. വീണ്ടും ശ്രമിക്കുന്നു > 0
3. enable.idempotence = true
4. max.in.flight.requests.per.connection ≤ 5 (1 ക്രമാനുഗതമായി അയയ്ക്കുന്നതിന്)
5. Transactional.id = ${application-name}-${hostname}

ഉപഭോക്താവ്:

1. isolation.level = read_committed

ഭാവിയിലെ ആപ്ലിക്കേഷനുകളിലെ പിശകുകൾ കുറയ്ക്കുന്നതിന്, സ്പ്രിംഗ് കോൺഫിഗറേഷനിൽ ഞങ്ങൾ ഞങ്ങളുടെ സ്വന്തം റാപ്പർ ഉണ്ടാക്കി, അവിടെ ലിസ്റ്റുചെയ്ത ചില പാരാമീറ്ററുകൾക്കുള്ള മൂല്യങ്ങൾ ഇതിനകം സജ്ജീകരിച്ചിരിക്കുന്നു.

സ്വയം പഠനത്തിനുള്ള രണ്ട് മെറ്റീരിയലുകൾ ഇതാ:

• KIP-98 - കൃത്യമായി ഒരിക്കൽ ഡെലിവറി, ഇടപാട് സന്ദേശമയയ്ക്കൽ
• ക്രമീകരണങ്ങളുടെ വിവരണം

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