Что такое Kafka backpressure?

Kafka Backpressure

Backpressure (встречное давление) в контексте Apache Kafka — это механизм контроля скорости обработки сообщений. Когда консьюмер не может обработать сообщения быстрее, чем они поступают, он применяет backpressure, сигнализируя продьюсеру замедлить отправку. Это критически важный концепт для стабильности высоконагруженных систем.

Проблема без Backpressure

Продьюсер                     Консьюмер
 (много сообщений/сек)      (медленная обработка)

  1000 msg/sec ---->
                            Буфер переполняется!
  1000 msg/sec ---->
                            OutOfMemory!
  1000 msg/sec ---->
                            Потеря данных!

Без backpressure буфер консьюмера переполняется, приводя к:

• OutOfMemory исключениям
• Потере сообщений
• Зависанию приложения

Как Kafka реализует Backpressure

Kafka имеет встроенный механизм backpressure через параметр fetch.max.bytes и обработку в консьюмере:

// Kafka ConsumerConfig с параметрами для контроля backpressure
public class KafkaConsumerConfig {
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        
        // Максимум 52MB данных за раз (default: 52428800)
        // Если консьюмер медленный, он берет меньше
        configProps.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 52428800);
        
        // Максимум 500ms ждать перед возвратом данных
        configProps.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1);
        configProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
        
        // Паузирование партиций если буфер полный
        configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); // Берем макс 500 записей за раз
        
        return new DefaultKafkaConsumerFactory<>(configProps);
    }
}

Механизм Backpressure на уровне Консьюмера

Решение 1: Ручная пауза при перегрузке

@Component
public class BackpressureConsumer {
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    @KafkaListener(topics = "orders", groupId = "order-processing")
    public void consumeOrders(List<ConsumerRecord<String, String>> records) {
        for (ConsumerRecord<String, String> record : records) {
            try {
                // Обрабатываем сообщение
                processOrder(record.value());
            } catch (ResourceExhaustedException e) {
                // Если нет ресурсов, применяем backpressure
                logger.warn("Resource exhausted, applying backpressure");
                throw e; // Консьюмер приостановится и переполучит сообщение позже
            }
        }
    }
    
    private void processOrder(String orderData) {
        // Обработка заказа
    }
}

Решение 2: Асинхронная обработка с очередью (рекомендуется)

@Component
public class BackpressureConsumerWithQueue {
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
    private final Queue<String> processingQueue = new LinkedBlockingQueue<>(1000); // Лимит на 1000
    
    @KafkaListener(topics = "orders", groupId = "order-processing")
    public void consumeOrders(String message) {
        // Если очередь полная, Kafka применяет backpressure
        // и консьюмер паузируется
        if (!processingQueue.offer(message)) {
            logger.warn("Processing queue full, applying backpressure");
            throw new BackpressureException("Queue overflow");
        }
    }
    
    @PostConstruct
    public void startProcessing() {
        executorService.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    String message = processingQueue.poll(1, TimeUnit.SECONDS);
                    if (message != null) {
                        processOrder(message);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });
    }
    
    private void processOrder(String orderData) {
        // Может занимать время
        // Тем временем консьюмер читает в очередь
    }
}

Backpressure на уровне Продьюсера

Продьюсер тоже может испытывать backpressure от Kafka брокера:

@Service
public class KafkaProducerWithBackpressure {
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final SemaphoreGate sendGate = new SemaphoreGate(1000); // Макс 1000 in-flight запросов
    
    public void sendOrder(String orderId, String orderData) {
        // Если in-flight запросов много, применяем backpressure
        if (!sendGate.tryAcquire()) {
            logger.warn("Backpressure: too many in-flight requests");
            // Можем либо выбросить exception, либо дождаться
            sendGate.acquire(); // Блокируем до освобождения слота
        }
        
        try {
            // Отправляем сообщение
            ListenableFuture<SendResult<String, String>> future = 
                kafkaTemplate.send("orders", orderId, orderData);
            
            future.addCallback(
                result -> {
                    logger.info("Message sent: " + orderId);
                    sendGate.release(); // Освобождаем слот
                },
                ex -> {
                    logger.error("Failed to send message", ex);
                    sendGate.release(); // Освобождаем слот даже при ошибке
                }
            );
        } catch (Exception e) {
            sendGate.release();
            throw e;
        }
    }
}

// Простая реализация Semaphore gate
public class SemaphoreGate {
    private final Semaphore semaphore;
    
    public SemaphoreGate(int permits) {
        this.semaphore = new Semaphore(permits);
    }
    
    public boolean tryAcquire() {
        return semaphore.tryAcquire();
    }
    
    public void acquire() throws InterruptedException {
        semaphore.acquire();
    }
    
    public void release() {
        semaphore.release();
    }
}

Пример: система обработки заказов с backpressure

@Service
public class OrderProcessingService {
    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private final OrderRepository orderRepository;
    private final PaymentService paymentService;
    private final NotificationService notificationService;
    
    // Ограничиваем одновременно обрабатываемые заказы
    private final Semaphore processingLimiter = new Semaphore(50);
    
    @KafkaListener(topics = "orders.created", groupId = "order-processor")
    public void onOrderCreated(OrderCreatedEvent event) throws InterruptedException {
        // Применяем backpressure если слишком много обработок
        if (!processingLimiter.tryAcquire()) {
            logger.warn("Too many orders being processed, applying backpressure");
            throw new BackpressureException("Processing limit reached");
        }
        
        try {
            processOrderAsync(event);
        } catch (Exception e) {
            processingLimiter.release();
            throw e;
        }
    }
    
    private void processOrderAsync(OrderCreatedEvent event) {
        // Обработка в отдельном потоке
        CompletableFuture.runAsync(() -> {
            try {
                Order order = orderRepository.findById(event.getOrderId());
                
                // Медленная обработка
                PaymentResult result = paymentService.charge(order);
                if (result.isSuccess()) {
                    order.markAsPaid();
                    orderRepository.save(order);
                    notificationService.sendConfirmation(order);
                }
            } finally {
                processingLimiter.release();
            }
        });
    }
}

Параметры для контроля Backpressure

Мониторинг Backpressure

@Component
public class BackpressureMonitor {
    private final MeterRegistry meterRegistry;
    private final AtomicInteger processingCount = new AtomicInteger(0);
    private final AtomicInteger backpressureEvents = new AtomicInteger(0);
    
    @PostConstruct
    public void setupMetrics() {
        meterRegistry.gauge("kafka.processing.count", processingCount);
        meterRegistry.counter("kafka.backpressure.events", backpressureEvents);
    }
    
    public void recordBackpressure() {
        backpressureEvents.incrementAndGet();
        // Альерт если слишком много backpressure событий
        if (backpressureEvents.get() > 100) {
            logger.error("High backpressure: " + backpressureEvents.get() + " events");
        }
    }
}

Лучшие практики

1. Используй асинхронную обработку - обработка в отдельных потоках
2. Ограничь количество in-flight запросов - используй Semaphore
3. Мониторь метрики - follow обработку и backpressure
4. Правильно настрой буфер - max.poll.records соответствующие памяти
5. Обрабатывай исключения - правильно освобождай ресурсы

Заключение

Backpressure в Kafka — это критический механизм для построения надежных систем обработки сообщений. Правильное использование гарантирует, что система не перегружается и может обрабатывать нагрузку стабильно, даже если есть пики активности.