רעפּראָסעססעד געשעענישן באקומען פון קאַפקאַ

העלא, האבר.

לעצטנס איך שערד זיין דערפאַרונג וועגן וואָס פּאַראַמעטערס מיר ווי אַ מאַנשאַפֿט רובֿ אָפט נוצן פֿאַר Kafka פּראָדוצירער און קאָנסומער צו באַקומען נעענטער צו געראַנטיד עקספּרעס. אין דעם אַרטיקל איך ווילן צו זאָגן איר ווי מיר אָרגאַניזירט די שייַעך-פּראַסעסינג פון אַ געשעעניש באקומען פון Kafka ווי אַ רעזולטאַט פון די צייטווייליגע אַנאַוויילאַביליטי פון די פונדרויסנדיק סיסטעם.

מאָדערן אַפּלאַקיישאַנז אַרבעטן אין אַ זייער קאָמפּליצירט סוויווע. געשעפט לאָגיק אלנגעוויקלט אין אַ מאָדערן טעכנאָלאָגיע אָנלייגן, פליסנדיק אין אַ דאָקקער בילד געראטן דורך אַן אָרקעסטראַטאָר ווי Kubernetes אָדער OpenShift, און קאַמיונאַקייטינג מיט אנדערע אַפּלאַקיישאַנז אָדער פאַרנעמונג סאַלושאַנז דורך אַ קייט פון גשמיות און ווירטועל ראָוטערס. אין אַזאַ אַ סוויווע, עפּעס קענען שטענדיק ברעכן, אַזוי ריפּראַסעסינג געשעענישן אויב איינער פון די פונדרויסנדיק סיסטעמען איז אַנאַוויילאַבאַל איז אַ וויכטיק טייל פון אונדזער געשעפט פּראַסעסאַז.

ווי עס איז געווען איידער קאַפקאַ

פריער אין די פּרויעקט, מיר געוויינט IBM MQ פֿאַר ייסינגקראַנאַס אָנזאָג עקספּרעס. אויב קיין טעות איז פארגעקומען בעשאַס די אָפּעראַציע פון ​​​​די סערוויס, די באקומען אָנזאָג קען זיין געשטעלט אין אַ טויט-בריוו-ריי (DLQ) פֿאַר ווייַטער מאַנואַל פּאַרסינג. די DLQ איז געווען באשאפן ווייַטער צו די ינקאַמינג ריי, דער אָנזאָג איז טראַנספערד אין IBM MQ.

אויב דער טעות איז געווען צייַטווייַליק און מיר קען באַשליסן עס (למשל, אַ ResourceAccessException אויף אַ הטטפּ רופן אָדער אַ MongoTimeoutException אויף אַ MongoDb בעטן), די ריטריי סטראַטעגיע וואָלט נעמען ווירקונג. רעגאַרדלעסס פון די בראַנטשינג לאָגיק פון די אַפּלאַקיישאַן, דער אָריגינעל אָנזאָג איז אריבערגעפארן אָדער צו די סיסטעם ריי פֿאַר דילייד שיקט, אָדער צו אַ באַזונדער אַפּלאַקיישאַן וואָס איז געווען געמאכט לאַנג צוריק צו שיקן אַרטיקלען ווידער. דאָס כולל אַ ריסענד נומער אין די אָנזאָג כעדער, וואָס איז טייד צו די פאַרהאַלטן מעהאַלעך אָדער די סוף פון די אַפּלאַקיישאַן-מדרגה סטראַטעגיע. אויב מיר האָבן ריטשט די סוף פון די סטראַטעגיע אָבער די פונדרויסנדיק סיסטעם איז נאָך אַנאַוויילאַבאַל, דער אָנזאָג וועט זיין געשטעלט אין די DLQ פֿאַר מאַנואַל פּאַרסינג.

געפֿינען אַ לייזונג

זוכן אויף די אינטערנעט, איר קענען געפֿינען די פאלגענדע באַשלוס. אין קורץ, עס איז פארגעלייגט צו שאַפֿן אַ טעמע פֿאַר יעדער פאַרהאַלטן מעהאַלעך און ינסטרומענט קאָנסומער אַפּלאַקיישאַנז אויף די זייַט, וואָס וועט לייענען אַרטיקלען מיט די פארלאנגט פאַרהאַלטן.

טראָץ די גרויס נומער פון positive באריכטן, עס מיינט צו מיר נישט גאָר געראָטן. ערשטער פון אַלע, ווייַל די דעוועלאָפּער, אין אַדישאַן צו ימפּלאַמענינג געשעפט רעקווירעמענץ, וועט האָבן צו פאַרברענגען אַ פּלאַץ פון צייַט ימפּלאַמענינג די דיסקרייבד מעקאַניזאַם.

אין אַדישאַן, אויב אַקסעס קאָנטראָל איז ענייבאַלד אויף די Kafka קנויל, איר וועט האָבן צו פאַרברענגען עטלעכע מאָל צו שאַפֿן טעמעס און צושטעלן די נייטיק אַקסעס צו זיי. אין אַדישאַן צו דעם, איר וועט דאַרפֿן צו אויסקלייַבן דעם ריכטיק רעטענטיאָן.מס פּאַראַמעטער פֿאַר יעדער פון די רעסאַפּיז טעמעס אַזוי אַז אַרטיקלען האָבן צייט צו פאַרדריסן און טאָן ניט פאַרשווינדן פון עס. די ימפּלאַמענטיישאַן און בעטן פון אַקסעס וועט זיין ריפּיטיד פֿאַר יעדער יגזיסטינג אָדער נייַ דינסט.

לאָמיר איצט זען וואָס מעקאַניזאַמז פרילינג אין אַלגעמיין און פרילינג-קאַפקאַ אין באַזונדער צושטעלן אונדז פֿאַר אָנזאָג ריפּראַסעסינג. Spring-kafka האט אַ טראַנזיטיוו דעפּענדענסי אויף פרילינג-פֿרייַ, וואָס גיט אַבסטראַקציעס פֿאַר אָנפירונג פאַרשידענע BackOffPolicies. דאָס איז אַ פערלי פלעקסאַבאַל געצייַג, אָבער זיין באַטייטיק שטערונג איז סטאָרינג אַרטיקלען פֿאַר ריסענדינג אין אַפּלאַקיישאַן זכּרון. דעם מיטל אַז ריסטאַרטינג די אַפּלאַקיישאַן רעכט צו אַ דערהייַנטיקן אָדער אַ אַפּעריישאַנאַל טעות וועט רעזולטאַט אין די אָנווער פון אַלע אַרטיקלען פּענדינג ריפּראַסעסינג. זינט דעם פונט איז קריטיש פֿאַר אונדזער סיסטעם, מיר האָבן נישט באַטראַכטן עס ווייַטער.

ספּרינג-קאַפקאַ זיך גיט עטלעכע ימפּלאַמאַנץ פון ContainerAwareErrorHandler, למשל 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
    }

אין דעם בייַשפּיל, retryAt איז די צייט צו ריסטאַרט די MessageListenerContainer אויב עס איז נאָך פליסנדיק. די שייַעך-קאַטער וועט פּאַסירן אין אַ באַזונדער פאָדעם לאָנטשט אין TaskScheduler, די ימפּלאַמענטיישאַן פון וואָס איז אויך צוגעשטעלט דורך פרילינג.

מיר געפֿינען די retryat ווערט אין די פאלגענדע וועג:

1. די ווערט פון די שייַעך-רופן טאָמבאַנק איז געקוקט אַרויף.
2. באַזירט אויף די טאָמבאַנק ווערט, די קראַנט פאַרהאַלטן מעהאַלעך אין די ריטריי סטראַטעגיע איז געזוכט. די סטראַטעגיע איז דערקלערט אין די אַפּלאַקיישאַן זיך; מיר אויסדערוויילט די JSON פֿאָרמאַט צו קראָם עס.
3. דער מעהאַלעך געפֿונען אין די JSON מענגע כּולל די נומער פון סעקונדעס נאָך וואָס פּראַסעסינג וועט זיין ריפּיטיד. די נומער פון סעקונדעס איז מוסיף צו די קראַנט צייט צו פאָרעם די ווערט פֿאַר retryAt.
4. אויב די מעהאַלעך איז ניט געפֿונען, די ווערט פון retryAt איז נאַל און דער אָנזאָג וועט זיין געשיקט צו DLQ פֿאַר מאַנואַל פּאַרסינג.

מיט דעם צוגאַנג, אַלע וואָס בלייבט איז צו ראַטעווען די נומער פון ריפּיטיד רופט פֿאַר יעדער אָנזאָג וואָס איז איצט פּראַסעסט, למשל אין אַפּלאַקיישאַן זכּרון. בעכעסקעם די ריטרישאַן ציילן אין זכּרון איז נישט קריטיש פֿאַר דעם צוגאַנג, ווייַל אַ לינעאַר לאָגיק אַפּלאַקיישאַן קען נישט שעפּן די פּראַסעסינג ווי אַ גאַנץ. ניט ענלעך פרילינג-פריי, ריסטאַרטינג די אַפּלאַקיישאַן וועט נישט פאַרשאַפן אַלע אַרטיקלען צו זיין פאַרפאַלן צו זיין ריפּראָסעסט, אָבער וועט פשוט ריסטאַרט די סטראַטעגיע.

דער צוגאַנג העלפט נעמען די מאַסע פון ​​די פונדרויסנדיק סיסטעם, וואָס קען זיין אַנאַוויילאַבאַל רעכט צו אַ זייער שווער מאַסע. אין אנדערע ווערטער, אין אַדישאַן צו ריפּראַסעסינג, מיר אַטשיווד די ימפּלאַמענטיישאַן פון דעם מוסטער קרייַז צוימען.

אין אונדזער פאַל, דער טעות שוועל איז בלויז 1, און צו מינאַמייז סיסטעם דאַונטיים רעכט צו צייטווייליגע נעץ אַוטידזשיז, מיר נוצן אַ זייער גראַניאַלער ריטריי סטראַטעגיע מיט קליין לייטאַנסי ינטערוואַלז. דאָס קען נישט זיין פּאַסיק פֿאַר אַלע גרופּע אַפּלאַקיישאַנז, אַזוי די שייכות צווישן די טעות שוועל און די ינטערוואַל ווערט מוזן זיין אויסגעקליבן באזירט אויף די קעראַקטעריסטיקס פון די סיסטעם.

א באַזונדער אַפּלאַקיישאַן פֿאַר פּראַסעסינג אַרטיקלען פון אַפּלאַקיישאַנז מיט ניט-דיטערמאַניסטיק לאָגיק

דאָ איז אַ ביישפּיל פון קאָד וואָס סענדז אַ אָנזאָג צו אַזאַ אַ אַפּלאַקיישאַן (Retryer), וואָס וועט רעסענד צו די דעסטינאַטיאָן טעמע ווען די 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);
    }

דער בייַשפּיל ווייזט אַז אַ פּלאַץ פון אינפֿאָרמאַציע איז טראַנסמיטטעד אין כעדערז. די ווערט פון RETRY_AT איז געפֿונען אין די זעלבע וועג ווי פֿאַר די ריטריי מעקאַניזאַם דורך די קאָנסומער האַלטן. אין אַדישאַן צו DESTINATION און RETRY_AT מיר פאָרן:

• GROUP_ID, דורך וואָס מיר גרופּע אַרטיקלען פֿאַר מאַנואַל אַנאַליסיס און סימפּלאַפייד זוכן.
• ORIGINAL_PARTITION צו פּרובירן צו האַלטן די זעלבע קאָנסומער פֿאַר שייַעך-פּראַסעסינג. דער פּאַראַמעטער קענען זיין נאַל, אין וואָס פאַל די נייַע צעטיילונג וועט זיין באקומען מיט די רעקאָרד.קיי () שליסל פון דער אָריגינעל אָנזאָג.
• דערהייַנטיקט די COUNTER ווערט צו נאָכפאָלגן די סטראַטעגיע ווידער.
• SEND_TO איז אַ קעסיידערדיק ינדאַקייטינג צי דער אָנזאָג איז געשיקט פֿאַר ריפּראַסעסינג ווען ריטשינג RETRY_AT אָדער געשטעלט אין DLQ.
• REASON - די סיבה וואָס אָנזאָג פּראַסעסינג איז ינטעראַפּטיד.

רעטריער סטאָרז אַרטיקלען פֿאַר ריסענדינג און מאַנואַל פּאַרסינג אין PostgreSQL. א טייַמער סטאַרץ אַ אַרבעט וואָס געפינט אַרטיקלען מיט RETRY_AT און סענדז זיי צוריק צו די ORIGINAL_PARTITION צעטיילונג פון די דעסטינאַטיאָן טעמע מיט די שליסל רעקאָרד.קיי ().

אַמאָל געשיקט, אַרטיקלען זענען אויסגעמעקט פֿון PostgreSQL. מאַנואַל פּאַרסינג פון אַרטיקלען אַקערז אין אַ פּשוט וי וואָס ינטעראַקץ מיט Retryer דורך REST API. די הויפּט פֿעיִקייטן זענען רעסענדינג אָדער ויסמעקן אַרטיקלען פון DLQ, וויוינג טעות אינפֿאָרמאַציע און זוכן פֿאַר אַרטיקלען, למשל דורך טעות נאָמען.

זינט אַקסעס קאָנטראָל איז ענייבאַלד אויף אונדזער קלאַסטערז, עס איז נייטיק צו אַדישנאַלי בעטן אַקסעס צו די טעמע וואָס Retrier איז צוגעהערט צו, און לאָזן Retrier צו שרייַבן צו די דעסטינאַטיאָן טעמע. דאָס איז ומבאַקוועם, אָבער ניט ענלעך די ינטערוואַל טעמע צוגאַנג, מיר האָבן אַ פול-פלעדזשד DLQ און וי צו פירן עס.

עס זענען קאַסעס ווען אַ ינקאַמינג טעמע איז לייענען דורך עטלעכע פאַרשידענע קאַנסומער גרופּעס, וועמענס אַפּלאַקיישאַנז ינסטרומענט פאַרשידענע לאָגיק. ריפּראַסעסינג אַ אָנזאָג דורך Retryer פֿאַר איינער פון די אַפּלאַקיישאַנז וועט רעזולטאַט אין אַ דופּליקאַט אויף די אנדערע. צו באַשיצן קעגן דעם, מיר מאַכן אַ באַזונדער טעמע פֿאַר שייַעך-פּראַסעסינג. די ינקאַמינג און ריטריי טעמעס קענען זיין לייענען דורך דער זעלביקער קאָנסומער אָן קיין ריסטריקשאַנז.

דורך פעליקייַט דעם צוגאַנג טוט נישט צושטעלן קרייַז צוימען פאַנגקשאַנאַליטי, אָבער עס קענען זיין מוסיף צו די אַפּלאַקיישאַן פרילינג-וואָלקן-נעטפליקס אָדער נייַ פרילינג וואָלקן קרייַז צוימען, ראַפּינג די ערטער ווו פונדרויסנדיק באַדינונגען זענען גערופן אין צונעמען אַבסטראַקציעס. אין דערצו, עס איז מעגלעך צו קלייַבן אַ סטראַטעגיע פֿאַר באַלקכעד מוסטער, וואָס קען אויך זיין נוציק. פֿאַר בייַשפּיל, אין פרילינג-וואָלקן-נעטפליקס דאָס קען זיין אַ פאָדעם בעקן אָדער אַ סעמאַפאָר.

רעזולטאַט

ווי אַ רעזולטאַט, מיר האָבן אַ באַזונדער אַפּלאַקיישאַן אַז אַלאַוז אונדז צו איבערחזרן אָנזאָג פּראַסעסינג אויב קיין פונדרויסנדיק סיסטעם איז טעמפּערעראַלי אַנאַוויילאַבאַל.

איינער פון די הויפּט אַדוואַנטידזשיז פון די אַפּלאַקיישאַן איז אַז עס קענען זיין געוויינט דורך פונדרויסנדיק סיסטעמען פליסנדיק אויף דער זעלביקער Kafka קנויל, אָן באַטייַטיק מאָדיפיקאַטיאָנס אויף זייער זייַט! אַזאַ אַ אַפּלאַקיישאַן וועט נאָר דאַרפֿן צו אַקסעס די ריטרייער טעמע, פּלאָמבירן אַ ביסל Kafka כעדערז און שיקן אַ אָנזאָג צו די Retrier. עס איז ניט דאַרפֿן צו כאַפּן קיין נאָך ינפראַסטראַקטשער. און אין סדר צו רעדוצירן די נומער פון אַרטיקלען טראַנספערד פון די אַפּלאַקיישאַן צו Retryer און צוריק, מיר יידענאַפייד אַפּלאַקיישאַנז מיט לינעאַר לאָגיק און שייַעך-פּראַסעסט זיי דורך די קאָנסומער האַלטן.

מקור: www.habr.com