За счет чего достигается отказоустойчивасть в Kafka

За счет чего достигается отказоустойчивость в Kafka

Kafka достигает отказоустойчивости через несколько архитектурных решений. Подробно объясню каждый механизм.

1. Репликация (Replication) — основной механизм

Kafka хранит каждый partition (раздел топика) на нескольких brokers одновременно:

Топик "orders" с 3 партициями и factor replication = 2:

Broker 1 (Leader)        Broker 2 (Replica)
┌─────────────────┐      ┌─────────────────┐
│ Partition 0     │◄────►│ Partition 0     │ (ISR: in-sync replicas)
│ (0, 100, 200)   │      │ (0, 100, 200)   │
└─────────────────┘      └─────────────────┘

Broker 2 (Leader)        Broker 3 (Replica)
┌─────────────────┐      ┌─────────────────┐
│ Partition 1     │◄────►│ Partition 1     │
│ (101, 201, 301) │      │ (101, 201, 301) │
└─────────────────┘      └─────────────────┘

Broker 3 (Leader)        Broker 1 (Replica)
┌─────────────────┐      ┌─────────────────┐
│ Partition 2     │◄────►│ Partition 2     │
│ (102, 202, 302) │      │ (102, 202, 302) │
└─────────────────┘      └─────────────────┘

Как это работает:

// Конфиг топика с репликацией
KafkaAdmin.createTopics(Collections.singleton(
    new NewTopic(
        "orders",        // имя топика
        3,              // количество партиций
        (short) 2       // replication factor (на 2 брокерах)
    )
)).all().get();

Что происходит при отказе:

Sценарий 1: Leader Broker 1 отказывает

ДО:
 Broker 1 (Leader) → Broker 2 (Replica, In-Sync)
         ↓
         X (отказал)

ПОСЛЕ (auto failover через 30 сек):
 Broker 2 (новый Leader, был Replica, In-Sync)
         ↑
    Никаких потерь данных!

2. ISR (In-Sync Replicas) — гарантия консистентности

ISR — это replicas, которые:

• Синхронизированы с leader
• Отстают максимум на replica.lag.time.max.ms миллисекунд
• Готовы стать leader

Конфиг Kafka для управления ISR:

server.properties:
min.insync.replicas=2        # Минимум replicas для ack
replica.lag.time.max.ms=10000  # 10 сек отставания
unclean.leader.election.enable=false  # НЕ выбирай replica с потерей данных

Producer отправляет сообщение:

// Настройка acks (сколько replicas должны подтвердить запись)
ProducerConfig config = new ProducerConfig();

// acks=1: Leader получил → OK (может быть потеря при отказе leader)
config.put("acks", 1);

// acks=all (или -1): Leader + ВСЕ ISR получили → OK (нет потерь)
config.put("acks", "all");

// acks=0: Не ждём подтверждения (самый быстрый, но ненадёжный)
config.put("acks", 0);

Практический пример:

public class KafkaProducerReliable {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        // Высокая надёжность
        props.put("acks", "all");  // Ждём подтверждения от всех ISR
        props.put("retries", 3);   // Повторяй при ошибке
        props.put("min.insync.replicas", 2); // Минимум 2 replicas должны получить
        
        // Производительность
        props.put("compression.type", "snappy");
        props.put("batch.size", 16384);
        props.put("linger.ms", 10);
        
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        
        try {
            String message = "Order created: 12345";
            ProducerRecord<String, String> record = 
                new ProducerRecord<>("orders", "order-12345", message);
            
            // Отправляем с гарантией доставки
            RecordMetadata metadata = producer.send(record).get();
            
            System.out.println("Message sent to partition " + 
                metadata.partition() + " at offset " + metadata.offset());
        } catch (Exception e) {
            System.err.println("Failed to send message: " + e.getMessage());
        } finally {
            producer.close();
        }
    }
}

3. Зооkeeper (Zookeeper Quorum) — координатор отказоустойчивости

Zookeeper следит за health brokers и координирует failover:

Zookeeper Quorum (3 ноды, нужна majority):
    ZK1 (Leader)    ZK2 (Follower)    ZK3 (Follower)
        ↓                  ↓                  ↓
    Consensus: если больше половины живо → сервис работает

Процесс автоматического failover:

1. Kafka Broker отправляет heartbeat в ZK каждые 6 секунд

2. ZK перестаёт получать heartbeat от Broker 1
   timeout = 30 сек → Broker 1 объявлен мёртвым

3. ZK триггерит Controller
   → "Broker 1 отказал, перевыбираем leader для его партиций"

4. Для каждой партиции где Broker 1 был leader:
   → Выбираем первого живого ISR члена
   → Назначаем его новым leader
   → Обновляем metadata

5. Все producers и consumers получают новую metadata
   → Переключаются на новый leader

Время отказоустойчивости: 30-40 сек

4. Распределённое хранение (Distributed Architecture)

Кластер из 5 Kafka brokers:

┌─────────────┐  ┌─────────────┐  ┌─────────────┐
│  Broker 1   │  │  Broker 2   │  │  Broker 3   │
│  Topics:    │  │  Topics:    │  │  Topics:    │
│  ├─ P0(L)   │  │  ├─ P0(R)   │  │  ├─ P1(L)   │
│  ├─ P1(R)   │  │  ├─ P1(R)   │  │  ├─ P1(R)   │
│  └─ P2(R)   │  │  └─ P2(L)   │  │  ├─ P2(R)   │
└─────────────┘  └─────────────┘  └─────────────┘
      ↑                ↑                ↑
   (L=Leader)      (R=Replica)    Каждый broker
                                  может отказать
   Если Broker 1 отказывает:
   - P0 leader переходит на Broker 2
   - P1 leader остаётся на Broker 2
   - P2 leader переходит на Broker 3

5. Durability — персистентность данных

Каждый broker хранит сообщения на диске с гарантией:

// Конфигурация для максимальной надёжности
server.properties:

# Синхронная запись на диск (медленнее, но надёжнее)
log.flush.interval.messages=1  # Флешить каждое сообщение

ИЛИ

# Асинхронная запись (быстрее, небольшой риск потери при краше)
log.flush.interval.messages=1000  # Флешить каждые 1000 сообщений
log.flush.interval.ms=1000        # Или каждую секунду

# Удаление старых логов
log.retention.hours=168  # Хранить 7 дней
log.segment.bytes=1073741824  # Размер segment: 1 GB

Физическое хранилище на диске:

/kafka-logs/orders-0/
├─ 00000000000000000000.log    # Segment 1: сообщения 0-100
├─ 00000000000000000100.log    # Segment 2: сообщения 100-200
├─ 00000000000000000200.log    # Segment 3: сообщения 200-300
├─ 00000000000000000000.index  # Index для быстрого поиска
└─ 00000000000000000000.timeindex

6. Consumer Group Coordination

Consumers работают в группах с балансированием и отказоустойчивостью:

public class KafkaConsumerResilient {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
        props.put("group.id", "order-processors");  // Consumer Group
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        
        // Отказоустойчивость consumer
        props.put("session.timeout.ms", 30000);  // 30 сек на heartbeat
        props.put("heartbeat.interval.ms", 10000); // heartbeat каждые 10 сек
        props.put("max.poll.interval.ms", 300000); // 5 минут на обработку
        
        // Offset management
        props.put("auto.offset.reset", "earliest");  // С начала, если нет offset
        props.put("enable.auto.commit", false);  // Ручной commit для надёжности
        
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("orders"));
        
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        // Обработка сообщения
                        System.out.println("Processing: " + record.value());
                        
                        // Ручное подтверждение
                        consumer.commitSync();
                    } catch (Exception e) {
                        // Ошибка обработки → нет commit → переобработаем позже
                        System.err.println("Failed to process: " + e.getMessage());
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }
}

Сценарий отказоустойчивости consumer:

Consumer Group: [Consumer1, Consumer2, Consumer3]
Topics: orders (3 партиции)

Сценарий 1: Consumer1 отказывает

ДО:
  Consumer1 обрабатывает Partition 0
  Consumer2 обрабатывает Partition 1
  Consumer3 обрабатывает Partition 2

ПОСЛЕ (rebalancing 30 сек):
  Consumer2 обрабатывает Partition 0 и 1
  Consumer3 обрабатывает Partition 2
  
Оффсеты сохранены в Kafka → нет потери данных

7. Практический пример: надёжный Kafka setup

public class ResilientKafkaSetup {
    // Конфиг для production
    public static Properties getProducerConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
        props.put("acks", "all");  // Все replicas подтверждают
        props.put("retries", Integer.MAX_VALUE);  // Бесконечные retry
        props.put("max.in.flight.requests.per.connection", 5);
        props.put("compression.type", "snappy");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return props;
    }
    
    public static Properties getConsumerConfig(String groupId) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka1:9092,kafka2:9092,kafka3:9092");
        props.put("group.id", groupId);
        props.put("enable.auto.commit", false);  // Ручной commit
        props.put("auto.offset.reset", "earliest");  // От начала
        props.put("session.timeout.ms", 30000);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }
}

Сводка механизмов отказоустойчивости Kafka

┌─────────────────────────────────────────────────────────────┐
│ 1. Репликация (Replication Factor)                         │
│    → Данные на нескольких brokers одновременно             │
├─────────────────────────────────────────────────────────────┤
│ 2. ISR (In-Sync Replicas)                                  │
│    → Только синхронизированные replicas участвуют в failover│
├─────────────────────────────────────────────────────────────┤
│ 3. Zookeeper Quorum                                         │
│    → Координирует failover (выбирает нового leader)        │
├─────────────────────────────────────────────────────────────┤
│ 4. Distributed Architecture                                │
│    → Партиции распределены по разным brokers               │
├─────────────────────────────────────────────────────────────┤
│ 5. Persistent Storage (диск)                               │
│    → Данные сохраняются на диск с гарантией               │
├─────────────────────────────────────────────────────────────┤
│ 6. Consumer Group Coordination                             │
│    → Автоматический rebalancing при отказе consumer       │
├─────────────────────────────────────────────────────────────┤
│ 7. Offset Management                                        │
│    → Offsets хранятся в Kafka, не теряются при отказе    │
└─────────────────────────────────────────────────────────────┘

Вывод

Отказоустойчивость Kafka достигается через:

1. Репликацию данных на несколько brokers
2. ISR для гарантии консистентности
3. Zookeeper для координации failover
4. Распределённую архитектуру для избежания single point of failure
5. Персистентность на диск
6. Автоматический failover при отказе broker или consumer

В результате Kafka гарантирует доставку сообщений (at-least-once) и может пережить отказ любого одного компонента в кластере.