Какие знаешь решения проблем при падении микросервиса при чтении из Kafka?

Решения проблем при падении микросервиса при чтении из Kafka

Получение сообщений из Kafka — критичный процесс в асинхронных системах. Падение микросервиса может привести к потере данных, дублированию сообщений или зависаниям. Рассмотрю надёжные решения для обработки различных сценариев отказа.

Типы проблем при обработке Kafka

1. Микросервис падает перед обработкой сообщения

Проблема: сообщение прочитано, но не обработано, сервис упал — сообщение потеряно.

// Плохо: auto-commit включён
@KafkaListener(topics = "orders", groupId = "order-service")
public void handleOrder(Order order) {
    // Если тут упадет - сообщение уже коммичено, потеряется
    orderService.process(order);
}

Решение 1: Manual commit (ручной коммит)

@KafkaListener(topics = "orders", groupId = "order-service")
public void handleOrder(Order order, Acknowledgment acknowledgment) {
    try {
        orderService.process(order);
        // Коммитим только после успешной обработки
        acknowledgment.acknowledge();
    } catch (Exception e) {
        // Если ошибка - не коммитим, сообщение вернётся в очередь
        logger.error("Failed to process order", e);
        throw e; // Выбросить исключение для переобработки
    }
}

Решение 2: Spring Kafka с retry

@Configuration
public class KafkaListenerConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> kafkaListenerContainerFactory(
            ConsumerFactory<String, Order> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, Order> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setCommonErrorHandler(
            new DefaultErrorHandler(
                new FixedBackOff(1000, 3)  // Retry 3 раза с интервалом 1 сек
            )
        );
        return factory;
    }
}

@KafkaListener(topics = "orders", groupId = "order-service")
public void handleOrder(Order order) {
    orderService.process(order);
    // При ошибке автоматически перепопытается 3 раза
}

2. Микросервис падает во время обработки

Проблема: сообщение в процессе обработки, сервис упал, обработка не завершена.

// Решение: идемпотентная обработка
@KafkaListener(topics = "payments", groupId = "payment-service")
public void handlePayment(Payment payment, Acknowledgment acknowledgment) {
    try {
        // Проверяем, не обработано ли уже
        if (paymentRepository.existsByExternalId(payment.getExternalId())) {
            acknowledgment.acknowledge();
            return;
        }
        
        // Обрабатываем платеж
        Payment processed = paymentService.process(payment);
        paymentRepository.save(processed);
        
        acknowledgment.acknowledge();
    } catch (Exception e) {
        logger.error("Failed to process payment: {}", payment.getExternalId(), e);
        throw e;
    }
}

Улучшенная архитектура с Dead Letter Topic

Dead Letter Topic (DLT) для поврежденных сообщений

Проблема: сообщение постоянно вызывает ошибку (невалидные данные, отсутствующий ресурс).

@Configuration
public class KafkaConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> kafkaListenerContainerFactory(
            ConsumerFactory<String, Order> consumerFactory,
            KafkaTemplate<String, Order> template) {
        ConcurrentKafkaListenerContainerFactory<String, Order> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        
        // Отправляем поврежденные сообщения в DLT
        factory.setCommonErrorHandler(
            new DefaultErrorHandler(
                (record, exception) -> {
                    // Отправляем в Dead Letter Topic
                    template.send("orders-dlt", record.key(), (Order) record.value());
                },
                new FixedBackOff(1000, 3)
            )
        );
        return factory;
    }
}

// Обработчик для DLT - с логированием и уведомлением
@KafkaListener(topics = "orders-dlt", groupId = "order-service-dlt")
public void handleFailedOrder(@Payload Order order,
                              @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                              @Header(KafkaHeaders.EXCEPTION_MESSAGE) String exceptionMessage) {
    logger.error("Order processing failed permanently. Order: {}, Reason: {}", order.getId(), exceptionMessage);
    
    // Отправляем алерт
    alertService.sendAlert("Order processing failed", order);
    
    // Сохраняем в БД для ручного разбора
    failedOrderRepository.save(new FailedOrder(order, exceptionMessage));
}

Circuit Breaker для зависимых сервисов

Проблема: микросервис зависит от другого сервиса (БД, API), который медленный или недоступен.

@Configuration
public class CircuitBreakerConfig {
    @Bean
    public CircuitBreaker orderServiceCircuitBreaker() {
        CircuitBreakerConfig config = new CircuitBreakerConfig();
        config.registerHealthIndicator(true);
        config.waitDurationInOpenState(Duration.ofSeconds(30));
        config.failureRateThreshold(50.0f);
        config.slowCallRateThreshold(50.0f);
        config.slowCallDurationThreshold(Duration.ofSeconds(2));
        
        CircuitBreakerRegistry registry = CircuitBreakerRegistry.of(config);
        return registry.circuitBreaker("orderService");
    }
}

@KafkaListener(topics = "orders", groupId = "order-service")
public void handleOrder(Order order, Acknowledgment acknowledgment) {
    try {
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("orderService");
        
        // Обработка с защитой circuit breaker
        Supplier<Void> decorated = CircuitBreaker.decorateSupplier(
            circuitBreaker,
            () -> {
                orderService.process(order);
                return null;
            }
        );
        
        Try.ofSupplier(decorated).onSuccess(v -> acknowledgment.acknowledge());
    } catch (Exception e) {
        logger.error("Circuit breaker opened for order", e);
        throw e; // Вернём в очередь для переобработки
    }
}

Graceful Shutdown (корректное завершение)

Проблема: микросервис получает SIGTERM, но рубит обработку «на лету».

# application.yml
spring:
  kafka:
    listener:
      poll-timeout: 3000
      type: batch
      concurrency: 3
  shutdown:
    wait-for-tasks-to-complete-on-shutdown: true
  jpa:
    properties:
      hibernate:
        jdbc:
          batch_size: 20
          fetch_size: 50

@Configuration
public class KafkaShutdownConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Order> kafkaListenerContainerFactory(
            ConsumerFactory<String, Order> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, Order> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        // Дождёмся завершения текущей обработки перед выключением
        factory.getContainerProperties().setShutdownTimeout(30000);
        return factory;
    }
}

@Component
public class GracefulShutdownListener {
    @PreDestroy
    public void shutdown() {
        logger.info("Graceful shutdown initiated");
        // Приложение будет ждать, пока все обработчики завершат работу
    }
}

Мониторинг и алерты

@Configuration
public class MonitoringConfig {
    @Bean
    public MeterRegistry meterRegistry() {
        return new PrometheusMeterRegistry(PrometheusConfig.DEFAULT);
    }
}

@Component
public class KafkaConsumerMetrics {
    private final MeterRegistry meterRegistry;
    private final Counter processedCounter;
    private final Counter failedCounter;
    private final Timer processingTimer;
    
    public KafkaConsumerMetrics(MeterRegistry meterRegistry) {
        this.meterRegistry = meterRegistry;
        this.processedCounter = Counter.builder("kafka.processed.total")
            .register(meterRegistry);
        this.failedCounter = Counter.builder("kafka.failed.total")
            .register(meterRegistry);
        this.processingTimer = Timer.builder("kafka.processing.duration")
            .register(meterRegistry);
    }
    
    @KafkaListener(topics = "orders", groupId = "order-service")
    public void handleOrder(Order order, Acknowledgment acknowledgment) {
        processingTimer.record(() -> {
            try {
                orderService.process(order);
                processedCounter.increment();
                acknowledgment.acknowledge();
            } catch (Exception e) {
                failedCounter.increment();
                logger.error("Failed to process order: {}", order.getId(), e);
                throw e;
            }
        });
    }
}

Чеклист для надёжной обработки Kafka

• Manual commit — коммитим только после успешной обработки
• Retry logic — автоматические переопробки с backoff
• Idempotency — каждое сообщение можно обработать дважды безопасно
• Dead Letter Topic — маршрут для поврежденных сообщений
• Circuit Breaker — защита от сбоев зависимых сервисов
• Graceful Shutdown — завершение без потери сообщений
• Мониторинг — метрики обработки, алерты при ошибках
• Логирование — трассировка каждого сообщения
• Batch processing — группировка сообщений для оптимизации

В целом: offsets управление + retry + идемпотентность + DLT = надёжная обработка Kafka.