Kafka 클러스터가 전부 죽었는데 왜 500이 아니라 504를 응답했을까?

무슨 일이 있었나

작업을 위해 Kafka 클러스터 전체를 내렸다. 예상한 동작은 이랬다.

1. Kafka 전송이 실패한다
2. onErrorResume이 실패를 받아 이벤트를 로컬 파일에 쌓는다
3. Kafka가 복구되면 스케줄러가 파일을 읽어 재전송한다

실제로 벌어진 일은 달랐다.

• 클라이언트가 504 Gateway Timeout 수신
• 일부 메시지 유실!! (장애 발생)
• 유실된 메시지에 대해 애플리케이션 로그가 하나도 남지 않음

마지막 항목이 특히 이상했다. 실패했다면 log.error라도 찍혀야 하는데 아무것도 없었다.

문제의 코드는 이런 모양이었다.

return kafkaSender.send(Mono.just(senderRecord))
        .next()
        .flatMap(result -> processResult(result, event, eventJson, topicName))
        .onErrorResume(error -> handleKafkaError(event, eventJson, topicName, error));
        //             ^^^^^^^^^^^^^^^^^^^^^ 파일 fallback 은 여기서만 발동

타임아웃 계층 역전

설정을 보니 금방 눈에 띄는 게 있었다.

nginx proxy_read_timeout (60s)  <  delivery.timeout.ms (120s)

• producer.send()는 레코드를 accumulator에 넣고 즉시 반환
• 실제 대기는 백그라운드에서 일어나고, 그 상한이 delivery.timeout.ms
• 브로커가 전부 죽어 있으면 결과 콜백은 120초가 지난 후 리턴
• nginx는 60초에 포기

클라이언트가 504 응답을 받는건 이걸로 설명된다.

그런데 max.block.ms는 왜 아무런 영향을 끼치지 않았나

표를 보면 30초짜리 max.block.ms가 있다. 브로커가 다 죽었으니 여기서 먼저 끊겨야 할 것 같은데 그렇지 않았다. 이 값은 producer.send()가 실제로 블로킹될 때만 적용되는데, 블로킹되는 경우는 아래 두 가지 상황일 때이다.

1. 토픽 메타데이터가 없을 때
   • 이미 토픽 메타데이터를 위한 캐시가 존재한다면 갱신 실패하는 경우라도 기존 캐시를 삭제하지 않음. (아래 코드 참조)

   // Metadata.failedUpdate()
   public synchronized void failedUpdate(long now) {
       this.lastRefreshMs = now;
       this.attempts++;
       this.equivalentResponseCount = 0;
   }   // ← cache 를 건드리지 않는다
   

   • 캐시는 성공한 MetadataResponse를 처리할 때만 교체
   • 클러스터가 죽어 갱신이 계속 실패하면 기존 캐시가 무기한 유지
   • 따라서, waitOnMetadata는 대기 없이 즉시 반환

   // KafkaProducer.waitOnMetadata()
   Cluster cluster = metadata.fetch();
   metadata.add(topic, nowMs);                            // 토픽 만료 시각 갱신
   Integer partitionsCount = cluster.partitionCountForTopic(topic);
   if (partitionsCount != null && (partition == null || partition < partitionsCount))
       return new ClusterAndWaitTime(cluster, 0);        // 대기 0 으로 즉시 반환
   

2. buffer.memory가 가득 찼을 때
   • 당시 buffer 가 가득찬 상태가 아니었음.

결과적으로 max.block.ms는 한번도 발동하지 않았고 이번 장애 상황에서 아무런 영향도 끼치지 않았다.

cancel은 error가 아니다

nginx가 proxy_read_timeout에 도달하면 업스트림 커넥션을 닫는다. 그러면 Reactor Netty가 연결 종료를 감지하고 진행 중인 응답 구독을 취소(cancel) 한다.

이와 같은 경우 Reactor 체인 중 onErrorResume 단계로 진입하지 않고 그냥 종료가 된 이유는 뭘까?

Reactive Streams에서 cancel과 onError는 완전히 다른 신호다.

onErrorResume은 위에서 내려오는 onError를 받아 동작하는 연산자다. 
반면, cancel()은 아래에서 위로 올라가는 구독 해지 통보이지 에러가 아니다. 타입도, 방향도, 진입점도 다르다.

• cancel()은 위로 올라가며 각 연산자의 구독을 해지할 뿐, 아래로 내려오는 onError와는 만나지 않음
• 결과적으로 onErrorResume도, doOnError도, doOnSuccess도 전부 실행되지 않음

Reactor 소스로 확인

onErrorResume의 구현을 열어보면 명확하다. (reactor-core 3.7.12)

static final class ResumeSubscriber<T> extends Operators.MultiSubscriptionSubscriber<T, T> {

    final Function<? super Throwable, ? extends Publisher<? extends T>> nextFactory;

    @Override
    public void onError(Throwable t) {          // ← fallback 은 오직 여기 안에만 존재
        if (!second) {
            second = true;
            Publisher<? extends T> p;
            try {
                p = Operators.toFluxOrMono(Objects.requireNonNull(nextFactory.apply(t), ...));
                //                                                ^^^^^^^^^^^^^^^^^^^^
                //                                    fallback 이 호출되는 유일한 지점
            } catch (Throwable e) { ... }
            p.subscribe(this);
        } else {
            actual.onError(t);
        }
    }
}

cancel()은 부모 구현을 그대로 상속한다.

// Operators.MultiSubscriptionSubscriber
public void cancel() {
    if (!cancelled) {
        cancelled = true;
        drain();            // 상위 구독 해지만 수행
    }
}

cancel이 올라갈 때 각 연산자는 자기 상태를 바꾸며, 뒤늦게 (위로 부터)도착한 신호는 조용히 버리고 끝낸다.

flatMap(Operators.MonoSubscriber)의 처리가 대표적이다.

public void cancel() {
    O v = value;
    value = null;
    STATE.set(this, CANCELLED);   // CANCELLED = 4
    discard(v);
}

public final void complete(@Nullable O v) {
    for (; ; ) {
        int state = this.state;
        // if state is >= HAS_CANCELLED or bit zero is set (*_HAS_VALUE) case, return
        if ((state & ~HAS_REQUEST_NO_VALUE) != 0) {
            this.value = null;
            discard(v);
            return;               // 조용히 버리고 끝
        }
        ...

• Kafka의 Callback은 Reactive Streams 밖의 객체이기 때문에 이 플래그들을 전혀 모르는 상태 
• 120초(delivery.timeout.ms) 뒤 콜백은 정상적으로 호출되고, 값이 Reactor 연산자 경계에 닿는 순간 버려짐
• "Kafka가 무시한다"가 아니라 "Reactor가 받아주지 않는다" 

Reactive Streams Rule 2.8: A Subscriber MUST be prepared to receive one or more onNext signals after having called Subscription.cancel()

실제로 코드에선 실패가 onError로 오지도 않았다

조금 다른 얘기이긴 한데.. 실제 장애가 발생했던 코드에서는 close 시그널 발생하기 전 Kafka 실패가 발생했다고 하더라도 onError로 오지 않는 문제가 있었다.

 reactor-kafka의 SenderOptions에 stopOnError(false)가 설정되어 있었는데..

SenderOptions.<String, String>create(configProps)
        .stopOnError(false)    // 이 부분
        .scheduler(Schedulers.boundedElastic());

이 설정에서 reactor-kafka의 producer Callback은 아래와 같이 동작한다.

Callback callback = (metadata, exception) -> {
    if (state.get() == State.COMPLETE) return;

    if (exception != null) {
        firstException.compareAndSet(null, exception);
        if (senderOptions.stopOnError() || senderOptions.fatalException(exception)) {
            onError(exception);       // ← stopOnError(true) 일 때만 이 경로
            return;
        }
    }
    actual.onNext(new Response<>(metadata, exception, correlationMetadata));
    //     ^^^^^^ stopOnError(false) → 실패도 '값'으로 onNext 전달
};

참고로, fatalException의 정의도 확인해보면 아래와 같이 범위가 아주 좁다.

default boolean fatalException(@NonNull Throwable t) {
    return t instanceof AuthenticationException || t instanceof ProducerFencedException;
}

즉, stopOnError=false일 때 delivery timeout을 포함한 대부분의 전송 실패는 onError가 아니라 onNext(SenderResult{exception})로 도착한다.

대응 방안

타임아웃 계층 정렬

가장 간단하고 효과가 큰 조치로 아래와 같은 타임아웃 설정에 대한 변경을 고려한다.

제안 근거

• delivery.timeout.ms 는 linger.ms + request.timeout.ms 보다 크거나 같아야 함
  • 현재 linger.ms = 5ms
• max.block.ms를 줄이는 것은 큰 부작용이 없음
  • 버퍼 고갈 시 BufferExhaustedException 발생
  • 이를 통해 fallback 로직(파일 저장)을 정상적으로 수행하도록 유도
• 따라서..

  max.block.ms(5s) + delivery.timeout.ms(40s) < response-timeout-ms(60s)   // 여유 15초
  

public 체인에 .timeout() 추가

return kafkaSender.send(Mono.just(senderRecord))
        .next()
        .flatMap(result -> processResult(result, event, eventJson, topicName))
        .timeout(Duration.ofSeconds(40))                                          // 추가
        .onErrorResume(error -> handleKafkaError(event, eventJson, topicName, error));

이 timeout의 목적은 producer를 멈추는 것이 아니라 정지 상태를 onError로 바꿔 onErrorResume이 반응하게 만드는 것이다.

• .timeout()은 필수가 아니라 방어적 상한
• producer.sent()가 블로킹되어야 하고, buffer.memory가 고갈되는 다음과 같은 경우를 대비하기 위함

수용과 전달의 분리 (store-and-forward)

위 대응 방안들은 Nginx의 60초 timeout을 전제로 한다. 클라이언트 자체 timeout이 더 짧거나 사용자가 중단하면 여전히 취소가 발생할 수 있다.

이에 대한 대응 방안으로 클라이언트의 요청 처리와 데이터 저장을 분리하여 처리하는 것을 고려할 수 있다.

아래는 스테이징 구현 선택지이다.

정리

이번 장애를 통해 얻은 교훈들이다.

1. cancel은 onError가 아니다. 방향도 인터페이스도 다르다. onErrorResume으로 감싸 두었다고 모든 비정상 종료가 잡히는 게 아니다.
2. fallback을 "실패 경로"에만 걸어두면 취소에는 무력하다. 실패 처리와 취소 처리는 별개다. doFinally / doOnCancel이 취소를 볼 수 있는 유일한 훅이다.
3. 타임아웃 계층은 바깥이 안보다 커야 한다. 그리고 "안쪽"이 하나가 아닐 수 있다. Kafka producer만 해도 max.block.ms와 delivery.timeout.ms가 직렬로 붙는다.
4. producer.send()는 취소되지 않는다. 리액티브 체인을 취소해도 레코드는 accumulator에 남는다. 이건 유실이 아니라 중복 문제로 바뀐다는 뜻이다.
5. 관측되지 않는 실패가 가장 위험하다. Operators.onDiscard는 조용하다. doOnCancel 로깅 한 줄이 있었다면 원인 파악이 훨씬 빨랐을 것이다.