Aloha Habr.
ʻO wau nei e pili ana i nā ʻāpana a mākou e hoʻohana pinepine ai no Kafka Producer a me Consumer e hoʻokokoke kokoke i ka hoʻouna ʻana. Ma kēia ʻatikala makemake wau e haʻi iā ʻoe pehea mākou i hoʻonohonoho ai i ka hana hou ʻana o kahi hanana i loaʻa mai Kafka ma muli o ka loaʻa ʻole o ka ʻōnaehana o waho.
Hoʻohana ʻia nā noi hou i kahi ʻano paʻakikī loa. ʻO ka loiloi pāʻoihana i kāʻei ʻia i kahi ʻenehana ʻenehana hou, e holo ana i kahi kiʻi Docker i mālama ʻia e kahi orkestra e like me Kubernetes a i ʻole OpenShift, a kamaʻilio pū me nā noi ʻē aʻe a i ʻole nā hoʻonā ʻoihana ma o ke kaulahao o nā mea hoʻokele kino a virtual. I loko o ia ʻano kaiapuni, hiki i kekahi mea ke haki i nā manawa a pau, no laila ʻo ka hoʻoponopono hou ʻana i nā hanana inā ʻaʻole i loaʻa kekahi o nā ʻōnaehana waho he mea nui ia o kā mākou kaʻina ʻoihana.
Pehea ma mua o Kafka
Ma mua o ka papahana ua hoʻohana mākou iā IBM MQ no ka hoʻouna ʻana i nā memo asynchronous. Inā loaʻa kekahi hewa i ka wā o ka lawelawe ʻana, hiki ke waiho ʻia ka memo i loaʻa i loko o kahi make-letter-queue (DLQ) no ka hoʻopau lima hou. Ua hana ʻia ka DLQ e pili ana i ka laina e hiki mai ana, ua hoʻoili ʻia ka leka i loko o IBM MQ.
Inā he wā pōkole ka hewa a hiki iā mākou ke hoʻoholo (no ka laʻana, he ResourceAccessException ma kahi kelepona HTTP a i ʻole he MongoTimeoutException ma kahi noi MongoDb), a laila e hoʻokō ʻia ka hoʻolālā hoʻāʻo hou. Me ka manaʻo o ka manaʻo lālā o ka noi, ua hoʻoneʻe ʻia ka memo kumu i ka pila ʻōnaehana no ka lohi ʻana i ka hoʻouna ʻana, a i ʻole i kahi noi ʻokoʻa i hana ʻia i ka wā ma mua e hoʻouna hou i nā memo. Loaʻa kēia i kahi helu hoʻouna hou ma ke poʻo memo, i hoʻopaʻa ʻia i ka wā lohi a i ʻole ka hopena o ka hoʻolālā pae noi. Inā ua hiki mākou i ka hopena o ka hoʻolālā akā ʻaʻole hiki ke loaʻa ka ʻōnaehana o waho, e kau ʻia ka memo i ka DLQ no ka hoʻopau lima.
Ke ʻimi nei i kahi hopena
, hiki iā ʻoe ke loaʻa i kēia mau mea . I ka pōkole, ua manaʻo ʻia e hana i kahi kumuhana no kēlā me kēia manawa lohi a hoʻokō i nā noi Consumer ma ka ʻaoʻao, e heluhelu i nā memo me ka lohi i koi ʻia.
ʻOiai ka nui o nā loiloi maikaʻi, ʻike ʻia iaʻu ʻaʻole kūleʻa loa. ʻO ka mea mua, no ka mea, ʻo ka mea hoʻomohala, ma waho aʻe o ka hoʻokō ʻana i nā koi ʻoihana, pono e hoʻolilo i ka manawa nui e hoʻokō i ka mīkini i wehewehe ʻia.
Eia kekahi, inā hiki ke hoʻohana ʻia ka mana komo ma ka pūʻulu Kafka, pono ʻoe e hoʻolilo i kekahi manawa i ka hana ʻana i nā kumuhana a me ka hāʻawi ʻana i ke komo pono iā lākou. Ma waho aʻe o kēia, pono ʻoe e koho i ka ʻāpana retention.ms kūpono no kēlā me kēia kumuhana e hoʻāʻo hou i hiki i nā memo ke loaʻa ka manawa e hoʻopiʻi ʻia a ʻaʻole e nalowale. Pono e hana hou ʻia ka hoʻokō a me ke noi ʻana no kēlā me kēia lawelawe i loaʻa a i ʻole lawelawe hou.
E ʻike kākou i ke ʻano o ka punawai ma ka laulā a me ka spring-kafka e hāʻawi mai iā mākou no ka hoʻoponopono hou ʻana i nā leka. Loaʻa i ka Spring-kafka kahi hilinaʻi transitive i ka hoʻāʻo hou ʻana, e hāʻawi ana i nā abstractions no ka mālama ʻana i nā BackOffPolicies like ʻole. He mea hana maʻalahi kēia, akā ʻo kāna drawback koʻikoʻi ka mālama ʻana i nā memo no ka hoʻouna hou ʻana i ka hoʻomanaʻo noi. ʻO ke ʻano o ka hoʻomaka hou ʻana i ka noi ma muli o kahi hoʻopiʻi a i ʻole kahi hewa hana e hopena i ka nalowale o nā memo āpau e kali ana i ka hana hou ʻana. No ka mea he koʻikoʻi kēia wahi no kā mākou ʻōnaehana, ʻaʻole mākou i noʻonoʻo hou aku.
Hāʻawi ʻo spring-kafka i kekahi mau hoʻokō o ContainerAwareErrorHandler, no ka laʻana , hiki iā ʻoe ke hana i ka memo ma hope me ka hoʻololi ʻole ʻana i ka offset inā loaʻa kahi hewa. E hoʻomaka ana me ka mana o ka puna-kafka 2.3, ua hiki ke hoʻonohonoho i BackOffPolicy.
Hāʻawi kēia ala i nā memo i hoʻoponopono hou ʻia e ola i ka hoʻomaka hou ʻana o ka noi, akā ʻaʻohe hana DLQ. Ua koho mākou i kēia koho ma ka hoʻomaka ʻana o 2019, me ka manaʻoʻiʻo ʻaʻole pono ʻo DLQ (laki mākou a ʻaʻole pono mākou ma hope o kekahi mau mahina o ka hana ʻana i ka noi me kahi ʻōnaehana reprocessing). Nā hewa no ka manawa i hoʻopau ʻia ʻo SeekToCurrentErrorHandler. Ua paʻi ʻia nā hewa i koe i loko o ka log, i hopena i ka offset, a ua hoʻomau ʻia ka hana me ka leka aʻe.
Hoʻoholo hope
ʻO ka hoʻokō ʻana ma luna o SeekToCurrentErrorHandler ua paipai iā mākou e hoʻomohala i kā mākou hana ponoʻī no ka hoʻouna hou ʻana i nā memo.
ʻO ka mea mua, makemake mākou e hoʻohana i ka ʻike i loaʻa a hoʻonui iā ia ma muli o ka loiloi noi. No ka noi laina laina, ʻoi aku ka maikaʻi o ka hoʻōki ʻana i ka heluhelu ʻana i nā memo hou no kahi manawa pōkole i kuhikuhi ʻia e ka hoʻolālā hoʻāʻo hou. No nā noi ʻē aʻe, makemake wau e loaʻa kahi helu hoʻokahi e hoʻokō i ka hoʻolālā hoʻāʻo hou. Eia kekahi, pono e loaʻa i kēia helu hoʻokahi ka hana DLQ no nā ala ʻelua.
Pono e mālama ʻia ka hoʻolālā hoʻāʻo iā ia iho i loko o ka noi, nona ke kuleana no ka hoʻihoʻi ʻana i ka wā aʻe i ka wā i loaʻa ai kahi hewa pōkole.
Ke kāpae nei i ka mea kūʻai aku no kahi noi laina laina
I ka hana ʻana me ka puna-kafka, ʻano like ke ʻano o ke code e hoʻōki i ka Consumer:
public void pauseListenerContainer(MessageListenerContainer listenerContainer,
Instant retryAt) {
if (nonNull(retryAt) && listenerContainer.isRunning()) {
listenerContainer.stop();
taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
return;
}
// to DLQ
}I ka laʻana, ʻo retryAt ka manawa e hoʻomaka hou ai i ka MessageListenerContainer inā e holo mau ana. E hoʻomaka ka hoʻokuʻu hou ʻana i kahi pae ʻokoʻa i hoʻokuʻu ʻia ma TaskScheduler, ka hoʻokō ʻana i hāʻawi ʻia e ka puna.
ʻIke mākou i ka waiwai retryAt ma kēia ʻano:
- Nānā ʻia ka waiwai o ka helu kelepona hou.
- Ma muli o ke kumu kūʻai, ʻimi ʻia ka wā lohi o ka hoʻolālā hoʻāʻo hou. Ua haʻi ʻia ka hoʻolālā ma ka noi ponoʻī; ua koho mākou i ke ʻano JSON e mālama ai.
- Aia i loko o ka pūʻulu JSON ka helu o nā kekona ma hope o ka hana ʻana e hana hou ʻia. Hoʻohui ʻia kēia helu kekona i ka manawa o kēia manawa e hoʻokumu i ka waiwai no retryAt.
- Inā ʻaʻole ʻike ʻia ka wā, a laila, ʻaʻole i pau ka waiwai o ka retryAt a e hoʻouna ʻia ka memo i DLQ no ka hoʻopau lima.
Me kēia ala, ʻo nā mea a pau i koe e mālama i ka helu o nā kelepona mau ʻia no kēlā me kēia memo e hana ʻia nei, no ka laʻana i ka hoʻomanaʻo noi. ʻAʻole koʻikoʻi ka mālama ʻana i ka helu hoʻāʻo hou ʻana i ka hoʻomanaʻo no kēia ala, no ka mea, ʻaʻole hiki i kahi noi linear logic ke mālama i ka hana holoʻokoʻa. ʻAʻole e like me ka hoʻāʻo hou ʻana i ka punawai, ʻo ka hoʻomaka hou ʻana i ka noi ʻaʻole e nalowale nā memo a pau e hana hou ʻia, akā e hoʻomaka hou i ka hoʻolālā.
Kōkua kēia ala e lawe i ka ukana ma waho o ka ʻōnaehana waho, ʻaʻole hiki ke loaʻa ma muli o kahi haʻahaʻa kaumaha loa. ʻO ia hoʻi, ma waho aʻe o ka hoʻoponopono hou ʻana, ua hoʻokō mākou i ka hoʻokō ʻana o ke kumu .
I kā mākou hihia, ʻo ka paepae kuhi he 1 wale nō, a no ka hōʻemi ʻana i ka manawa haʻahaʻa o ka ʻōnaehana ma muli o ka pau ʻana o ka pūnaewele no ka manawa lōʻihi, hoʻohana mākou i kahi hoʻolālā hoʻāʻo nui loa me nā wā latency liʻiliʻi. ʻAʻole kūpono paha kēia no nā noi hui āpau, no laila pono e koho ʻia ka pilina ma waena o ka paepae hewa a me ka waiwai waena ma muli o nā ʻano o ka ʻōnaehana.
He palapala hoʻokaʻawale no ka hoʻoili ʻana i nā memo mai nā noi me ka loiloi non-deterministic
Eia kekahi laʻana o ke code e hoʻouna ana i ka memo i ia palapala noi (Retryer), e hoʻouna hou i ke kumuhana DESTINATION ke hiki i ka manawa RETRY_AT:
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);
}Hōʻike ka laʻana i ka nui o ka ʻike i hoʻouna ʻia ma nā poʻo. Loaʻa ka waiwai o RETRY_AT ma ke ʻano like me ka mīkini hoʻāʻo hou ma o ka Consumer stop. Ma waho aʻe o DESTINATION a me RETRY_AT, hele mākou:
- GROUP_ID, kahi e hui pū ai mākou i nā memo no ka nānā lima lima a me ka huli maʻalahi.
- ORIGINAL_PARTITION e ho'āʻo e mālama i ka mea kūʻai like no ka hana hou ʻana. He null paha kēia ʻāpana, a laila e loaʻa ka ʻāpana hou me ka hoʻohana ʻana i ke kī record.key() o ka memo kumu.
- Ua hōʻano hou ʻia ka waiwai COUNTER e hahai i ka hoʻolālā hoʻāʻo hou.
- ʻO SEND_TO kahi hōʻike mau inā hoʻouna ʻia ka memo no ka hoʻoponopono hou ʻana i ka hiki ʻana iā RETRY_AT a i ʻole ke waiho ʻia ma DLQ.
- REASON - ke kumu i hoʻopau ʻia ai ka hoʻoili ʻana i nā leka.
Mālama ʻo Retryer i nā memo no ka hoʻouna hou ʻana a me ka paʻi lima ma PostgreSQL. Hoʻomaka ka manawa i kahi hana e loaʻa ai nā memo me RETRY_AT a hoʻihoʻi iā lākou i ka ʻāpana ORIGINAL_PARTITION o ke kumuhana DESTINATION me ke kī record.key().
Ke hoʻouna ʻia, holoi ʻia nā memo mai PostgreSQL. Hana ʻia ka paʻi lima o nā memo ma kahi UI maʻalahi e launa pū me Retryer ma o REST API. ʻO kāna mau hiʻohiʻona nui ka hoʻouna ʻana a i ʻole ka holoi ʻana i nā memo mai DLQ, ke nānā ʻana i ka ʻike hewa a me ka ʻimi ʻana i nā memo, no ka laʻana ma ka inoa hewa.
No ka mea ua hiki ke hoomalu i ke komo ma ko kakou mau puulu, pono e noi hou aku i ke komo i ke kumuhana a Retryer e hoolohe nei, a e ae ia Retryer e kakau i ke kumuhana DESTINATION. He mea paʻakikī kēia, akā, ʻaʻole e like me ke ʻano o ke kumuhana interval, loaʻa iā mākou kahi DLQ piha a me UI e hoʻokele ai.
Aia kekahi mau hihia ke heluhelu ʻia kahi kumuhana e hiki mai ana e nā hui mea kūʻai like ʻole, nona nā noi e hoʻokō i nā loiloi like ʻole. ʻO ka hoʻoponopono hou ʻana i kahi memo ma o Retryer no kekahi o kēia mau noi e hopena i kahi kope ma kekahi. No ka pale ʻana i kēia, hana mākou i kumuhana ʻokoʻa no ka hana hou ʻana. Hiki ke heluhelu ʻia nā kumuhana e hiki mai ana a hoʻāʻo hou e ka mea kūʻai aku me ka ʻole o ka palena.
ʻAʻole hāʻawi kēia ʻano hana i ka hana hoʻopau kaapuni, akā hiki ke hoʻohui ʻia i ka noi me ka hoʻohana ʻana a mea hou paha , e kāʻei ana i nā wahi i kapa ʻia ai nā lawelawe waho i nā abstractions kūpono. Eia kekahi, hiki ke koho i kahi hoʻolālā no kumu hoʻohālike, hiki ke hoʻohana. No ka laʻana, i ka punawai-cloud-netflix hiki ke lilo i kolamu thread a i ʻole he semaphore.
hopena
ʻO ka hopena, loaʻa iā mākou kahi noi ʻokoʻa e hiki ai iā mākou ke hana hou i ka hoʻoili ʻana i ka leka inā loaʻa ʻole kekahi ʻōnaehana waho.
ʻO kekahi o nā pono nui o ka noi hiki ke hoʻohana ʻia e nā ʻōnaehana waho e holo ana ma ka pūʻulu Kafka like, me ka ʻole o nā hoʻololi koʻikoʻi ma ko lākou ʻaoʻao! ʻO ia palapala noi e komo wale i ke kumuhana hoʻāʻo hou, hoʻopiha i kekahi mau poʻomanaʻo Kafka a hoʻouna i kahi leka i ka Retryer. ʻAʻohe pono e hoʻonui i nā ʻōnaehana hou. A i mea e hōʻemi ai i ka helu o nā memo i hoʻoili ʻia mai ka noi iā Retryer a i hope, ua ʻike mākou i nā noi me ka loina laina a hoʻoponopono hou iā lākou ma o ka Consumer stop.
Source: www.habr.com