В какой ситуации сообщение брокера может не доставиться

Причины потери сообщений в брокере

Неуспешная доставка сообщения брокера зависит от архитектуры системы, конфигурации и обработки ошибок. Вот основные ситуации:

1. Потеря сообщений на стороне производителя

Асинхронная отправка без подтверждения:

# Без гарантий доставки
producer.send_async("topic", "message")
# Сообщение может быть потеряно если приложение крахнется

Правильно — ждём подтверждения:

# Гарантия «по крайней мере один раз»
future = producer.send("topic", "message")
record_metadata = future.get(timeout=10)  # Блокирующий вызов
print(f"Доставлено в partition {record_metadata.partition}")

2. Потеря на стороне брокера

Недостаточная репликация:

• Сообщение записано только на один узел
• Узел падает до репликации на остальные
• Решение: настрой min.insync.replicas >= 2, acks=all в Kafka

Неправильная конфигурация:

# Неправильно
producer = KafkaProducer(
    acks=1,
    retries=0
)

# Правильно
producer = KafkaProducer(
    acks='all',
    retries=3,
    max_in_flight_requests_per_connection=1
)

3. Потеря на стороне потребителя

Автоматический коммит смещения до обработки:

# Опасно — offset коммитится до обработки
consumer = KafkaConsumer(
    'topic',
    enable_auto_commit=True,
    auto_commit_interval_ms=1000
)

for message in consumer:
    try:
        process_message(message)
    except Exception:
        pass  # Сообщение потеряно!

Безопасно — ручной коммит после обработки:

consumer = KafkaConsumer(
    'topic',
    enable_auto_commit=False
)

for message in consumer:
    try:
        process_message(message)
        consumer.commit()
    except Exception as e:
        logger.error(f"Ошибка: {e}")
        # Смещение не коммитится — переобработаем

4. Сетевые проблемы

• Timeout при доставке
• Потеря пакетов
• Разрыв соединения

Решение: exponential backoff + retry:

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def send_with_retry(broker, message):
    return broker.send(message)

5. Poison pills (токсичные сообщения)

consumer = KafkaConsumer('topic')

for message in consumer:
    try:
        process_message(message)
        consumer.commit()
    except Exception:
        # Отправи в dead-letter queue вместо потери
        dead_letter_queue.send(message)
        consumer.commit()

Итоги

Ключевые гарантии:

• At most once: быстро, но теряет сообщения
• At least once: медленнее, но может дублировать
• Exactly once: гарантия уникальности

Для надёжной системы используй acks=all, ручной коммит, DLQ и мониторинг.