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

Механизмы обеспечения сохранности данных при отказе брокера в Kafka

Safety и reliability при потере брокера — критичны для production систем. Kafka имеет несколько встроенных механизмов, которые защищают данные от потери.

1. Репликация (Replication Factor)

Это основной механизм сохранности. Каждый партition реплицируется на несколько брокеров.

from kafka import KafkaProducer, KafkaConsumer
from kafka.admin import KafkaAdminClient, NewTopic

# Создаём топик с replication factor = 3
admin_client = KafkaAdminClient(bootstrap_servers='localhost:9092')

topic = NewTopic(
    name='my-topic',
    num_partitions=3,
    replication_factor=3,  # Данные хранятся на 3 брокерах
    topic_configs={
        'retention.ms': '604800000',  # 7 дней
        'min.insync.replicas': '2'    # Минимум 2 replica для acks
    }
)

admin_client.create_topics(new_topics=[topic])

# Если один из трёх брокеров отказывает:
# 1. Данные остаются на двух оставшихся брокерах
# 2. Kafka выбирает нового leader
# 3. Система продолжает работать

2. In-Sync Replicas (ISR)

ISR — это список реплик, которые полностью синхронизированы с leader. Гарантирует что данные на них актуальны.

from kafka import KafkaProducer

# Конфиг для producer — ожидает подтверждение от ISR реплик
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    acks='all',  # Или acks=-1: ждём подтверждения от ВСЕХ in-sync реплик
    retries=3,
    max_in_flight_requests_per_connection=1  # Сохраняем порядок
)

# Отправляем сообщение
future = producer.send('my-topic', b'important data')

# Блокируем до получения подтверждения
try:
    record_metadata = future.get(timeout=10)
    print(f'Сообщение отправлено на partition {record_metadata.partition}')
    print(f'Offset: {record_metadata.offset}')
except Exception as e:
    print(f'Ошибка: {e}')

producer.close()

Уровни acks для producer:

• acks=0 — не ждём подтверждения (опасно!)
• acks=1 — ждём подтверждения от leader
• acks=-1 или acks='all' — ждём от всех ISR (безопаснее)

3. min.insync.replicas (Минимум синхронных реплик)

Если число ISR упадёт ниже этого значения, producer получит ошибку.

# На уровне топика
from kafka.admin import ConfigResource, ConfigResourceType

config_resource = ConfigResource(
    ConfigResourceType.TOPIC,
    'my-topic',
    configs={'min.insync.replicas': '2'}
)

admin_client.alter_configs(config_resources=[config_resource])

# Producer с acks='all'
producer = KafkaProducer(
    bootstrap_servers='localhost:9092',
    acks='all'
)

# Сценарий:
# 1. У нас replication_factor=3, min.insync.replicas=2
# 2. Leader на брокере 1, follower на 2 и 3
# 3. Брокер 3 падает — ISR = [1, 2] (2 реплики) ✓ OK
# 4. Брокер 2 падает — ISR = [1] (1 реплика) ✗ Ошибка!
# 5. Producer не может отправить данные

4. Leader Election и Failover

Когда leader падает, Kafka автоматически выбирает нового leader из ISR.

# Мониторим брокеры
from kafka.admin import KafkaAdminClient
from kafka.structs import TopicPartition

admin_client = KafkaAdminClient(bootstrap_servers='localhost:9092')

# Получаем метаданные
metadata = admin_client.describe_cluster()
print(f'Брокеры: {metadata.brokers()}')

# Информация о партициях
from kafka.consumer.group import KafkaConsumer

consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers='localhost:9092'
)

# Получаем информацию о партициях
partitions = consumer.partitions_for_topic('my-topic')
for partition in partitions:
    tp = TopicPartition('my-topic', partition)
    consumer.assign([tp])
    metadata = consumer.highwater(tp)
    print(f'Partition {partition}: {metadata}')

5. Unclean Leader Election

Опасная опция — выбирает leader из не-ISR реплик, если все ISR недоступны. Может привести к потере данных!

# Конфиг брокера (kafka/config/server.properties)
# unclean.leader.election.enable=false  # Рекомендуется!

# Если false:
# - Брокер становится unavailable вместо выбора грязного leader
# - Гарантирует консистентность данных
# - Producer получает ошибку и может retry

# Если true:
# - Выбирается leader с наиболее свежими данными (может быть потеря!)
# - Система остаётся доступна

6. Дублирование обработки (Consumer Side)

В случае отказа consumer, другой consumer может обработать те же сообщения. Используй idempotent обработку.

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers='localhost:9092',
    group_id='my-group',
    enable_auto_commit=False,  # Ручной commit
    auto_offset_reset='earliest',
    session_timeout_ms=30000,  # Таймаут сессии
    heartbeat_interval_ms=10000
)

# Idempotent обработка
processed_messages = set()  # В production используй БД

for message in consumer:
    msg_id = f"{message.partition}_{message.offset}"
    
    # Проверяем что уже обработали
    if msg_id in processed_messages:
        print(f'Сообщение {msg_id} уже обработано, пропускаем')
        continue
    
    try:
        # Обработка
        data = json.loads(message.value.decode())
        print(f'Обработка: {data}')
        
        # Коммитим offset
        consumer.commit()
        processed_messages.add(msg_id)
    except Exception as e:
        print(f'Ошибка обработки: {e}')
        # Не коммитим — перепроцессируем при перезагрузке

7. Persistence и Durability на диске

Данные сохраняются на диск каждого брокера. Можешь настроить flush интервалы.

# Топик конфиг для максимальной durability
topic_config = {
    'flush.messages': '1000',  # Flush каждые 1000 сообщений
    'flush.ms': '1000',        # Или каждую 1 сек
    'log.cleanup.policy': 'delete',  # Удалять старые логи
    'retention.ms': '604800000',     # Хранить 7 дней
    'segment.ms': '86400000'         # Сегмент в день
}

# На диске: /var/kafka-logs/{topic}-{partition}/
# 00000000000000000000.log  <- основной лог
# 00000000000000000000.index <- индекс
# 00000000000000000000.timeindex <- временной индекс

8. Broker.rack конфигурация

Распределяет реплики по разным рэкам/датацентрам для избежания потери из-за отказа одного ДЦ.

# kafka/config/server.properties
broker.rack=us-east-1a

# Когда создаёшь топик с rack-aware
topic = NewTopic(
    name='critical-topic',
    num_partitions=6,
    replication_factor=3,  # На разные рэки
    topic_configs={}
)

# Kafka распределит реплики так:
# Partition 0: Broker 1 (rack a), Broker 2 (rack b), Broker 3 (rack c)
# Partition 1: Broker 2 (rack b), Broker 3 (rack c), Broker 1 (rack a)
# Partition 2: Broker 3 (rack c), Broker 1 (rack a), Broker 2 (rack b)

9. Monitoring и alerting

import time
from kafka.admin import KafkaAdminClient

class KafkaHealthMonitor:
    def __init__(self, bootstrap_servers):
        self.admin = KafkaAdminClient(bootstrap_servers=bootstrap_servers)
    
    def check_under_replicated_partitions(self):
        """Проверяет партиции с недостаточным числом реплик"""
        cluster = self.admin.describe_cluster()
        under_replicated = []
        
        for broker in cluster.brokers():
            # Ищем партиции где недостаточно ISR
            # Требует дополнительного мониторинга
            pass
        
        return under_replicated
    
    def check_broker_availability(self):
        """Проверяет доступность брокеров"""
        try:
            cluster = self.admin.describe_cluster()
            available_brokers = len(cluster.brokers())
            print(f'Доступно брокеров: {available_brokers}')
            return available_brokers
        except Exception as e:
            print(f'Ошибка подключения: {e}')
            return 0

monitor = KafkaHealthMonitor('localhost:9092')
available = monitor.check_broker_availability()
if available < 2:
    print('ALERT: Недостаточно брокеров!')

10. Рекомендации для production

• replication_factor >= 3 — минимум для critical топиков
• min.insync.replicas >= 2 — гарантирует потребление с 2+ синхронных реплик
• acks='all' в producer — ждём подтверждения от всех ISR
• enable_auto_commit=False в consumer — ручной контроль над commit
• unclean.leader.election.enable=false — не выбирай грязные leaders
• Мониторим ISR — alert если ISR < replication_factor
• Используй multiple racks/datacenters — защита от отказа ДЦ
• Тестируй failover сценарии — выключай брокеры и проверяй