Kafka에서 수신된 이벤트 재처리

헤이 하브르.

최근에 나는 자신의 경험을 공유했다 Kafka Producer와 Consumer가 보장된 전달에 더 가까워지도록 팀에서 가장 자주 사용하는 매개변수는 무엇입니까? 이 글에서는 외부 시스템을 일시적으로 사용할 수 없게 되어 Kafka에서 수신한 이벤트를 어떻게 다시 처리하는지 설명하고자 합니다.

최신 애플리케이션은 매우 복잡한 환경에서 작동합니다. Kubernetes나 OpenShift와 같은 오케스트레이터가 관리하는 Docker 이미지에서 실행되는 최신 기술 스택에 포함된 비즈니스 로직은 물리적 및 가상 라우터 체인을 통해 다른 애플리케이션이나 엔터프라이즈 솔루션과 통신합니다. 이러한 환경에서는 언제나 무언가가 고장날 수 있으므로 외부 시스템 중 하나를 사용할 수 없는 경우 이벤트를 다시 처리하는 것은 비즈니스 프로세스의 중요한 부분입니다.

카프카 이전의 상황은 어땠는가

이 프로젝트의 초기 단계에서는 비동기 메시지 전달을 위해 IBM MQ를 사용했습니다. 서비스 운영 중 오류가 발생하면 수신된 메시지는 추가 수동 구문 분석을 위해 데드 레터 큐(DLQ)에 보관될 수 있습니다. DLQ는 수신 대기열 옆에 생성되었으며, 메시지 전송은 IBM MQ 내부에서 발생했습니다.

오류가 일시적이고 이를 감지할 수 있는 경우(예: HTTP 호출에서 ResourceAccessException이 발생하거나 MongoDb 요청에서 MongoTimeoutException이 발생하는 경우) 재시도 전략이 적용됩니다. 애플리케이션 로직의 분기와 관계없이 원래 메시지는 지연 전송을 위해 시스템 대기열로 전송되거나, 메시지를 다시 전송하기 위해 만들어진 별도의 애플리케이션으로 전송되었습니다. 이 경우 재시도 번호가 메시지 헤더에 기록되며, 이는 지연 간격이나 애플리케이션 수준 전략의 끝에 연결됩니다. 전략의 끝에 도달했지만 외부 시스템을 여전히 사용할 수 없는 경우 해당 메시지는 수동 구문 분석을 위해 DLQ에 저장됩니다.

솔루션 검색

인터넷을 검색한 후, 다음을 찾을 수 있습니다 결정. 간단히 말해, 각 지연 간격에 대한 주제를 만들고, 필요한 지연 시간으로 메시지를 읽을 애플리케이션 측의 소비자를 구현하는 것이 제안됩니다.

긍정적인 리뷰가 많음에도 불구하고, 제 생각에는 전적으로 성공적이지는 않은 것 같습니다. 첫째, 개발자는 비즈니스 요구 사항을 구현하는 것 외에도 설명된 메커니즘을 구현하는 데 많은 시간을 할애해야 합니다.

또한, Kafka 클러스터에서 액세스 제어가 활성화된 경우 토픽을 생성하고 해당 토픽에 필요한 액세스 권한을 제공하는 데 시간을 할애해야 합니다. 이 외에도 각 검색 주제에 대해 올바른 retention.ms 매개변수를 선택해야 메시지가 다시 전송되고 사라지지 않을 시간을 확보할 수 있습니다. 구현 및 액세스 요청은 기존 서비스나 새로운 서비스마다 반복되어야 합니다.

이제 Spring이 제공하는 메시지 재처리 메커니즘과 특히 Spring-Kafka가 제공하는 메시지 재처리 메커니즘을 살펴보겠습니다. Spring-kafka는 다양한 BackOffPolicy를 관리하기 위한 추상화를 제공하는 spring-retry에 대한 전이적 종속성을 가지고 있습니다. 이것은 매우 유연한 도구이지만, 가장 큰 단점은 다시 보낼 메시지를 애플리케이션의 메모리에 저장한다는 것입니다. 즉, 업데이트나 작업 중 오류로 인해 애플리케이션을 다시 시작하면 재처리를 기다리는 모든 메시지가 손실됩니다. 이 지점은 우리 시스템에 매우 중요하므로 더 이상 고려하지 않았습니다.

spring-kafka 자체는 ContainerAwareErrorHandler의 여러 구현을 제공합니다. 예를 들어 SeekToCurrentErrorHandler이를 통해 오류가 발생한 경우 오프셋을 이동하지 않고도 나중에 메시지를 처리할 수 있습니다. spring-kafka 2.3부터 BackOffPolicy를 설정할 수 있습니다.

이 접근 방식을 사용하면 재처리된 메시지가 애플리케이션을 다시 시작한 후에도 유지되지만 여전히 DLQ 메커니즘은 없습니다. 이것은 우리가 2019년 초에 선택한 옵션으로, DLQ가 필요하지 않을 것이라고 낙관적으로 믿었습니다(우리는 운이 좋았고 실제로 그러한 재처리 시스템이 있는 애플리케이션을 사용한 몇 달 동안 DLQ가 필요하지 않았습니다). 일시적인 오류로 인해 SeekToCurrentErrorHandler가 트리거되었습니다. 나머지 오류는 로그에 인쇄되어 오프셋이 발생했으며, 처리는 다음 메시지로 계속되었습니다.

최종 결정

SeekToCurrentErrorHandler 기반 구현으로 인해 우리는 메시지 재전송을 위한 자체 메커니즘을 개발하게 되었습니다.

우선, 우리는 기존 경험을 활용하고 애플리케이션의 논리에 따라 이를 확장하고 싶었습니다. 선형 논리를 사용하는 애플리케이션의 경우 재시도 전략에서 지정한 짧은 시간 내에 새 메시지 읽기를 중지하는 것이 가장 좋습니다. 나머지 애플리케이션의 경우 재시도 전략을 시행할 수 있는 단일 지점을 원했습니다. 또한, 이 단일 지점에는 두 가지 접근 방식 모두에 대한 DLQ 기능이 있어야 합니다.

재시도 전략 자체는 일시적인 오류가 발생했을 때 다음 간격을 얻는 역할을 하는 애플리케이션에 저장되어야 합니다.

선형 논리 애플리케이션의 소비자 중지

spring-kafka를 사용할 때 Consumer를 중지하는 코드는 다음과 같습니다.

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에서 시작된 별도의 스레드에서 발생하며, 이에 대한 구현도 Spring에서 제공됩니다.

retryAt의 값은 다음과 같은 방법으로 찾을 수 있습니다.

1. 콜백 카운터의 값을 검색하고 있습니다.
2. 카운터 값에 따라 재시도 전략의 현재 지연 간격이 검색됩니다. 전략은 애플리케이션 자체에서 선언되며, 이를 저장하기 위해 JSON 형식을 선택했습니다.
3. JSON 배열에서 찾은 간격에는 처리를 반복해야 하는 초 수가 포함되어 있습니다. 이 초 수는 현재 시간에 더해져 retryAt 값을 형성합니다.
4. 간격을 찾을 수 없는 경우 retryAt 값은 null이고 메시지는 수동 구문 분석을 위해 DLQ로 전송됩니다.

이 접근 방식을 사용하면 현재 처리 중인 각 메시지에 대한 반복 호출 횟수를 애플리케이션 메모리 등에 저장하는 것만 남습니다. 이 접근 방식에서는 시도 카운터를 메모리에 보관하는 것이 중요하지 않습니다. 선형 논리를 사용하는 애플리케이션은 전체적으로 처리를 수행할 수 없기 때문입니다. spring-retry와 달리 애플리케이션을 다시 시작해도 재처리할 메시지가 모두 손실되지 않고 전략만 다시 시작됩니다.

이러한 접근 방식은 매우 무거운 부하로 인해 사용할 수 없는 외부 시스템의 부하를 덜어주는 데 도움이 됩니다. 즉, 재처리 외에도 패턴 구현까지 달성했습니다. 회로 차단기.

우리의 경우 오류 임계값은 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을 사용하여 재처리를 위해 동일한 Consumer를 유지하려고 합니다. 이 매개변수는 null일 수 있으며, 이 경우 새 파티션은 원본 메시지의 record.key() 키로 얻어집니다.
• 재시도 전략에 따라 COUNTER 값이 업데이트되었습니다.
• SEND_TO는 RETRY_AT에 도달하면 재처리를 위해 메시지를 보낼지 아니면 DLQ에 넣을지 여부를 나타내는 상수입니다.
• 이유 - 메시지 처리가 중단된 이유.

Retryer는 PostgreSQL에서 재전송 및 수동 구문 분석을 위해 메시지를 저장합니다. 타이머는 RETRY_AT가 발생한 메시지를 찾고 record.key() 키와 함께 DESTINATION 토픽의 ORIGINAL_PARTITION 파티션으로 다시 보내는 작업을 시작합니다.

전송된 메시지는 PostgreSQL에서 삭제됩니다. 메시지의 수동 구문 분석은 REST API를 통해 Retryer와 상호 작용하는 간단한 UI에서 이루어집니다. 주요 기능으로는 DLQ에서 메시지를 다시 보내거나 삭제하고, 오류 정보를 보고, 오류 이름으로 메시지를 검색하는 것이 있습니다.

클러스터에서 액세스 제어가 활성화되었으므로 Retryer가 수신하는 주제에 대한 액세스를 추가로 요청하고 Retryer가 DESTINATION 주제에 쓸 수 있도록 허용해야 합니다. 이 방법은 불편하지만 주제별 간격 방식과 달리 본격적인 DLQ와 이를 관리하기 위한 UI를 얻을 수 있습니다.

여러 소비자 그룹이 들어오는 주제를 읽고 해당 그룹의 애플리케이션이 서로 다른 논리를 구현하는 경우가 있습니다. 이러한 애플리케이션 중 하나에서 Retryer를 통해 메시지를 다시 시도하면 다른 애플리케이션에서도 중복이 발생합니다. 이를 방지하기 위해 재처리를 위한 별도의 주제를 만들고 있습니다. 수신 및 재시도 주제는 동일한 소비자가 아무런 제한 없이 읽을 수 있습니다.

기본적으로 이 접근 방식은 회로 차단기 기능을 제공하지 않지만 다음을 사용하여 애플리케이션에 회로 차단기 기능을 추가할 수 있습니다. 스프링 클라우드 넷플릭스 또는 새로운 스프링 클라우드 회로 차단기외부 서비스의 호출 사이트를 적절한 추상화로 래핑합니다. 또한, 전략을 선택할 수 있는 기회가 있습니다. 칸막이 패턴도 유용할 수 있습니다. 예를 들어, spring-cloud-netflix에서는 스레드 풀이나 세마포어가 될 수 있습니다.

출력

그 결과, 외부 시스템을 일시적으로 사용할 수 없을 때 메시지 처리를 반복할 수 있는 별도의 애플리케이션을 얻게 되었습니다.

이 애플리케이션의 주요 장점 중 하나는 동일한 Kafka 클러스터에서 실행되는 외부 시스템에서도 큰 수정 없이 사용할 수 있다는 것입니다! 이러한 애플리케이션은 재시도 주제에 액세스하고, 몇 개의 Kafka 헤더를 작성하고, Retryer에 메시지를 보내기만 하면 됩니다. 추가적인 인프라를 구축할 필요가 없습니다. 그리고 애플리케이션에서 Retryer로 전송되는 메시지 수를 줄이기 위해 선형 논리를 적용한 애플리케이션을 선택하고 Consumer를 중지하여 해당 애플리케이션에서 재처리를 수행했습니다.

출처 : habr.com