Lagi retry Spring Kafka: jak naprawić

Nieblokujący retry w Spring Kafka: diagnozowanie opóźnień i rozwiązanie

Podczas implementacji asynchronicznego retry dla Kafka z użyciem @RetryableTopic w Spring Boot 3.2 i Spring Kafka 3.1 pojawiają się opóźnienia na retry-topicach pod obciążeniem. Główny topic jest przetwarzany równomiernie, ale retry-0 gromadzi wiadomości przez 10–12 minut, następnie częściowo je opróżnia i ponownie się zatyka. Prowadzi to do szarpnięć w przetwarzaniu.

Metryki pokazują rzadkie skoki fetch-requestów zamiast równomiernego poll. Consumer okresowo odpykuje brokera dla heartbeat, ale rzeczywiste odczyty następują sporadycznie z wysokim opóźnieniem.

Kluczowe metryki do analizy

Do diagnozy zbierz następujące metryki Spring Kafka:

• kafka_consumer_time_between_poll_avg / max — odstępy między poll.
• kafka_consumer_last_poll_seconds_ago — czas od ostatniego poll.
• kafka_consumer_poll_idle_ratio_avg — udział idle poll.
• kafka_consumer_fetch_manager_fetch_rate / fetch_latency_avg — prędkość i opóźnienia fetch.
• spring_kafka_listener_seconds_count / sum — czas przetwarzania listenera.

Wykresy fetch-rate i listener invocations potwierdzają: przetwarzanie aktywuje się szarpnięciami, poll idle jest wysoki.

Diagnoza przez threaddump

Threaddump wykazał ~180 wątków w TIMED_WAITING na ScheduledThreadPoolExecutor$DelayedWorkQueue.take. Wskazuje to na przeciążenie scheduler'a używanego do odłożonego resume partycji.

Mechanizm pauzy w RetryableTopic

@RetryableTopic stosuje ContainerPartitionPausingBackOffManager: przy pierwszym przetworzeniu retry-wiadomości sprawdza backoff, pauzuje partycję na now() + delay i planuje resume przez ListenerContainerPauseService.

Kod pauzy:

private final TaskScheduler scheduler;

public void pausePartition(MessageListenerContainer messageListenerContainer, 
                           TopicPartition partition,
                           Duration pauseDuration) {
    Instant resumeAt = Instant.now().plusMillis(pauseDuration.toMillis());
    messageListenerContainer.pausePartition(partition);
    this.scheduler.schedule(() -> {
        messageListenerContainer.resumePartition(partition);
    }, resumeAt);
}

Przy wysokim obciążeniu kolejka zadań scheduler'a przepełnia się. Domyślnie spring.task.scheduling.pool.size=1, co tworzy wąskie gardło: zadania na resume kumulują się, fetch nie uruchamia się na czas.

Rozwiązanie: zwiększenie puli scheduler

Ustaw spring.task.scheduling.pool.size=10 w application.properties.

Rezultaty:

• Opóźnienie retry-0 ustabilizowało się na akceptowalnym poziomie (nie spada do 0 z powodu backoff).
• Poll-interwały skróciły się, fetch-rate wzrósł.
• Listener invocations stały się równomierne.

Wykres lag po poprawce: utrzymuje się poziomo bez wzrostu. Fetch latency spadła, poll częstszy.

Co jest ważne

• Ograniczony scheduler: domyślna pula 1 wątku nie radzi sobie z masowymi pause/resume przy obciążeniu na retry.
• Metryki pierwszorzędne: monitoruj fetch-rate, poll-idle, listener-time przed threaddump.
• Transakcje drugorzędne: ack-mode=record, max.poll.records=1 nie wpływają na lag.
• Concurrency OK: Spring Kafka poprawnie skaluje według partycji, nawet z mnożeniem topiców.
• Alternatywa: dla highload rozważ custom retry z timestamp-check w pętli zamiast pauz.

Rekomendacje monitorowania

Dodaj dashboardy:

• Commit/rollback rate dla exactly-once.
• Queue size scheduler'a (custom metric).
• Partition pause/resume events.
• Backoff timestamp vs current time delta.

Testuj pod obciążeniem przed produkcją. Źródła Spring Kafka dają pełny obraz mechaniki.

— Editorial Team