Как гарантировать доставку сообщений в Kafka?

Гарантирование доставки сообщений в Apache Kafka

Кафка предоставляет несколько уровней гарантирования доставки сообщений. Рассмотрю детально механизмы, конфигурации и best practices.

Уровни гарантирования доставки (Delivery Semantics)

At Most Once (максимум один раз)

• Сообщение может быть потеряно
• Сообщение НИКОГДА не будет обработано более одного раза
• Самый быстрый, но ненадежный вариант
• Использование: логирование, метрики, где потеря данных допустима

At Least Once (минимум один раз)

• Сообщение гарантированно будет доставлено
• Сообщение может быть обработано несколько раз
• Требует idempotent обработки на стороне consumer
• Использование: большинство бизнес-критичных приложений

Exactly Once (ровно один раз)

• Сообщение доставлено РОВНО один раз
• Наиболее надежный и сложный в реализации
• Требует специальной конфигурации и Kafka 0.11+
• Использование: финансовые транзакции, критичные данные

At Least Once: конфигурация Producer

Конфигурация на стороне производителя

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    # At Least Once гарантия
    acks='all',  # Ждем подтверждение от ALL реплик
    retries=3,  # Повторяем при ошибке
    max_in_flight_requests_per_connection=1,  # Порядок сообщений
    compression_type='snappy',  # Сжатие
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

try:
    future = producer.send('orders', {'order_id': 123, 'amount': 99.99})
    record_metadata = future.get(timeout=10)  # Блокирующее ожидание
    print(f"Message sent to partition {record_metadata.partition} "
          f"at offset {record_metadata.offset}")
except Exception as e:
    print(f"Failed to send: {e}")
finally:
    producer.close()

Параметры Producer для надежности:

• acks='all' — ждать подтверждения от leader и всех in-sync реплик
• retries=3 (или больше) — повторять при ошибке
• max_in_flight_requests_per_connection=1 — обеспечивает порядок
• compression_type — сжатие уменьшает пропускную способность

Конфигурация Broker

Параметры Kafka broker (server.properties)

# Минимальное количество in-sync реплик
min.insync.replicas=2

# Время удержания логов
log.retention.hours=168  # 7 дней

# Репликация фактор (должен быть минимум 2)
default.replication.factor=3

# Количество broker-ов, которые должны быть активны
unclean.leader.election.enable=false

Критические параметры:

• min.insync.replicas=2 — гарантирует, что данные на 2+ репликах
• unclean.leader.election.enable=false — не выбирает лидера из non-in-sync реплик

At Least Once: конфигурация Consumer

Управление offset-ом

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processor',
    auto_offset_reset='earliest',  # Начать с начала при первом запуске
    enable_auto_commit=False,  # ВАЖНО: ручное управление offset-ом
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    max_poll_records=100,
    session_timeout_ms=30000
)

for message in consumer:
    try:
        # Обработка сообщения (должна быть идемпотентной)
        process_order(message.value)
        # ТОЛЬКО после успешной обработки коммитим offset
        consumer.commit(offsets={message.topic_partition: 
                                 message.offset + 1})
    except Exception as e:
        print(f"Failed to process: {e}")
        # При ошибке НЕ коммитим offset
        # Сообщение будет переобработано при перезагрузке

Ключевые моменты:

• enable_auto_commit=False — ручное управление offset-ом
• Коммитим только ПОСЛЕ успешной обработки
• При ошибке не коммитим, сообщение переобработается

Exactly Once: дополнительные меры

Использование Kafka транзакций (0.11+)

from kafka import KafkaProducer, KafkaConsumer

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    transactional_id='unique-id-1',  # ВАЖНО: уникальный ID для идемпотентности
    acks='all',
    retries=5,
    max_in_flight_requests_per_connection=1
)

consumer = KafkaConsumer(
    'input_topic',
    bootstrap_servers=['localhost:9092'],
    group_id='processor',
    isolation_level='read_committed'  # Читать только committed сообщения
)

for message in consumer:
    with producer.transaction():
        try:
            # Обработка
            result = process(message)
            # Отправка результата в другой топик
            producer.send('output_topic', result)
            # Коммит offset внутри транзакции
            consumer.commit()
        except Exception as e:
            # Откат всех операций
            raise

Параметры для Exactly Once:

• transactional_id — уникальный ID для идемпотентности
• isolation_level='read_committed' — читать только committed данные
• Используются Kafka транзакции

Дополнительные механизмы надежности

Idempotent обработка на стороне Consumer

class OrderProcessor:
    def __init__(self, db):
        self.db = db
    
    def process_order(self, order_data):
        order_id = order_data['order_id']
        
        # Проверка: может ли быть повторно обработано?
        if self.db.order_exists(order_id):
            return  # Уже обработано, ничего не делаем
        
        # Обработка
        self.db.insert_order(order_data)
        self.db.update_inventory(order_data)
        
        # Отправка события (с idempotent_id)
        return {'order_id': order_id, 'status': 'processed'}

Dead Letter Queue для необработанных сообщений

def process_with_dlq(consumer, producer):
    for message in consumer:
        try:
            process_order(message)
            consumer.commit()
        except Exception as e:
            # Отправить в DLQ с информацией об ошибке
            producer.send('orders-dlq', {
                'original_message': message.value,
                'error': str(e),
                'timestamp': datetime.now()
            })

Мониторинг и отладка

Проверка лага Consumer

# Смотреть лаг для группы
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group order-processor --describe

Проверка репликации

# Смотреть under-replicated партиции
kafka-topics --bootstrap-server localhost:9092 \
  --describe --under-replicated-partitions

Рекомендуемая конфигурация для production

Для At Least Once (рекомендуется для 95% случаев):

• Producer: acks='all', retries=5, max_in_flight_requests_per_connection=1
• Consumer: enable_auto_commit=False, ручное управление offset
• Broker: min.insync.replicas=2, replication.factor=3
• Приложение: реализовать idempotency

Для Exactly Once (финансовые транзакции):

• Использовать Kafka транзакции
• transactional_id на producer
• isolation_level='read_committed' на consumer
• Дополнительное логирование и мониторинг

Вывод

Доставка сообщений в Kafka — это не автоматическое явление, а результат правильной конфигурации. Комбинация параметров producer, consumer, broker и idempotent обработки гарантирует надежное сохранение и обработку данных даже в случае сбоев.