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

헤이 하브르.

최근에 나는 자신의 경험을 공유했다 Kafka 생산자 및 소비자가 보장된 전달에 더 가까워지기 위해 팀으로서 가장 자주 사용하는 매개변수에 대해 설명합니다. 이 기사에서는 외부 시스템을 일시적으로 사용할 수 없어 Kafka에서 수신한 이벤트의 재처리를 어떻게 구성했는지 설명하고 싶습니다.

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

카프카 이전에는 어땠나요?

프로젝트 초기에는 비동기식 메시지 전달을 위해 IBM MQ를 사용했습니다. 서비스 작업 중에 오류가 발생하면 수신된 메시지는 추가 수동 구문 분석을 위해 DLQ(배달 못한 편지 대기열)에 배치될 수 있습니다. DLQ는 수신 큐 옆에 생성되었으며 메시지는 IBM MQ 내부로 전송되었습니다.

오류가 일시적이고 이를 확인할 수 있는 경우(예: HTTP 호출의 ResourceAccessException 또는 MongoDb 요청의 MongoTimeoutException) 재시도 전략이 적용됩니다. 애플리케이션의 분기 논리에 관계없이 원래 메시지는 지연 전송을 위해 시스템 큐로 이동되거나 메시지를 재전송하기 위해 오래 전에 만들어진 별도의 애플리케이션으로 이동되었습니다. 여기에는 지연 간격 또는 애플리케이션 수준 전략의 끝과 연결된 메시지 헤더의 재전송 번호가 포함됩니다. 전략이 끝났지만 외부 시스템을 여전히 사용할 수 없는 경우 수동 구문 분석을 위해 메시지가 DLQ에 배치됩니다.

솔루션 검색

인터넷에서 검색하기, 다음을 찾을 수 있습니다 결정. 즉, 각 지연 간격에 대한 주제를 생성하고 필요한 지연 시간 동안 메시지를 읽는 소비자 애플리케이션을 측면에서 구현하는 것이 제안됩니다.

많은 긍정적인 리뷰에도 불구하고 완전히 성공하지는 못한 것 같습니다. 우선, 개발자는 비즈니스 요구 사항을 구현하는 것 외에도 설명된 메커니즘을 구현하는 데 많은 시간을 소비해야 하기 때문입니다.

또한 Kafka 클러스터에서 액세스 제어가 활성화된 경우 주제를 생성하고 필요한 액세스 권한을 제공하는 데 시간을 투자해야 합니다. 또한 메시지가 재전송될 시간을 갖고 메시지에서 사라지지 않도록 각 재시도 항목에 대해 올바른tention.ms 매개 변수를 선택해야 합니다. 기존 서비스나 신규 서비스 각각에 대해 액세스 구현 및 요청을 반복해야 합니다.

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

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

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

최종 결정

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

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

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

선형 논리 애플리케이션에 대한 소비자 중지

spring-kafka로 작업할 때 소비자를 중지하는 코드는 다음과 같습니다.

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은 재처리를 위해 동일한 소비자를 유지하려고 합니다. 이 매개변수는 null일 수 있으며, 이 경우 원본 메시지의 Record.key() 키를 사용하여 새 파티션을 가져옵니다.
• 재시도 전략을 따르도록 COUNTER 값을 업데이트했습니다.
• SEND_TO는 메시지가 RETRY_AT에 도달한 후 재처리를 위해 전송되는지 아니면 DLQ에 배치되는지를 나타내는 상수입니다.
• REASON - 메시지 처리가 중단된 이유입니다.

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로 전송되는 메시지 수를 줄이기 위해 선형 로직으로 애플리케이션을 식별하고 Consumer stop을 통해 재처리했습니다.

출처 : habr.com