Что такое репликация в Kafka?

Репликация в Kafka

Репликация в Kafka — это механизм обеспечения высокой доступности и отказоустойчивости сообщений путём создания копий данных каждого раздела (partition) на нескольких серверах (brokers) в кластере. Это позволяет гарантировать, что данные не будут потеряны при выходе из строя одного или нескольких узлов кластера.

Основные концепции

Репликация в Kafka основана на концепции лидер-последователь:

• Leader (лидер) — основной сервер, который обрабатывает все чтение и запись для раздела
• Followers (последователи) — сервера-реплики, которые синхронно копируют данные от лидера
• ISR (In-Sync Replicas) — набор реплик, которые полностью синхронизированы с лидером

Ключевые характеристики:

• Автоматическое резервное копирование данных
• Быстрый failover при падении лидера
• Настраиваемый уровень гарантий доставки
• Расчёт каждого раздела на отказ до N-1 узлов

Архитектура репликации

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.Future;

public class KafkaReplicationExample {
    public static void main(String[] args) throws Exception {
        // Конфигурация продюсера с гарантиями репликации
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "broker1:9092,broker2:9092,broker3:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, 
            StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
            StringSerializer.class.getName());
        
        // ACKS=ALL - подтверждение только после репликации на все ISR
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        
        // Минимум реплик, которые должны подтвердить запись
        props.put("min.insync.replicas", 2);
        
        // Количество попыток переотправки при ошибке
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        
        KafkaProducer<String, String> producer = 
            new KafkaProducer<>(props);
        
        try {
            // Отправка сообщения с гарантией репликации
            ProducerRecord<String, String> record = 
                new ProducerRecord<>("my-topic", "key", "value");
            
            Future<RecordMetadata> future = producer.send(record, 
                (metadata, exception) -> {
                    if (exception != null) {
                        System.out.println("Ошибка: " + exception.getMessage());
                    } else {
                        System.out.println("Сообщение отправлено на раздел: " + 
                            metadata.partition() + ", смещение: " + 
                            metadata.offset() + ", лидер: " + 
                            metadata.brokerId());
                    }
                });
            
            RecordMetadata metadata = future.get();
        } finally {
            producer.close();
        }
    }
}

Уровни гарантий (ACKs)

1. acks=0 (Без подтверждения)

props.put(ProducerConfig.ACKS_CONFIG, "0");
// Продюсер не ждёт подтверждения
// Высокая скорость, но риск потери данных
// Использование: логирование, метрики

2. acks=1 (Подтверждение лидера)

props.put(ProducerConfig.ACKS_CONFIG, "1");
// Лидер подтверждает получение, но не реплики
// Средний баланс скорость/надёжность
// Риск потери при падении лидера до репликации

3. acks=all (Подтверждение всех реплик)

props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put("min.insync.replicas", 2);
// Наиболее надёжный уровень
// Гарантирует отсутствие потерь данных
// Использование: финансовые, критичные системы

Управление репликацией

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.TopicPartition;

public class ReplicationManagement {
    public static void main(String[] args) throws Exception {
        Properties adminProps = new Properties();
        adminProps.put("bootstrap.servers", 
            "broker1:9092,broker2:9092,broker3:9092");
        
        AdminClient adminClient = AdminClient.create(adminProps);
        
        try {
            // Получение информации о топике
            var topicDescFuture = adminClient.describeTopics(
                java.util.Collections.singletonList("my-topic"));
            var topicDesc = topicDescFuture.get().get("my-topic");
            
            // Анализ реплик
            topicDesc.partitions().forEach(partition -> {
                System.out.println("Раздел: " + partition.partition());
                System.out.println("Лидер: " + partition.leader().id());
                System.out.println("Реплики: " + 
                    partition.replicas().stream()
                        .map(n -> n.id())
                        .toList());
                System.out.println("ISR: " + 
                    partition.isr().stream()
                        .map(n -> n.id())
                        .toList());
            });
        } finally {
            adminClient.close();
        }
    }
}

Failover и переизбрание лидера

public class ReplicationFailover {
    /**
     * Когда лидер падает:
     * 1. Контроллер Kafka обнаруживает падение
     * 2. Выбирает новый лидер из ISR (первый в списке)
     * 3. Обновляет метаданные
     * 4. Продюсеры и консьюмеры переподключаются
     */
    
    public static void simulateFailover() {
        System.out.println("Начальная конфигурация:");
        System.out.println("Раздел 0: Leader=1, Replicas=[1,2,3], ISR=[1,2,3]");
        System.out.println();
        
        System.out.println("Брокер 1 падает...");
        System.out.println();
        
        System.out.println("Новая конфигурация:");
        System.out.println("Раздел 0: Leader=2, Replicas=[1,2,3], ISR=[2,3]");
        System.out.println("Брокер 1 остаётся в списке реплик, но не в ISR");
    }
}

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

# server.properties

# Коэффициент репликации (default: 3)
default.replication.factor=3

# Минимум синхронизированных реплик для ack=all
min.insync.replicas=2

# Время ожидания от последователя перед удалением из ISR (ms)
replica.lag.time.max.ms=10000

# Размер буфера для репликации
replica.socket.receive.buffer.bytes=65536

# Управление удалением старых логов
log.retention.hours=168
log.retention.bytes=1073741824

Мониторинг репликации

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;

public class ReplicationMonitoring {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, 
            "broker1:9092,broker2:9092,broker3:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "monitoring-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, 
            StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, 
            StringDeserializer.class.getName());
        
        // Метрики консьюмера включены по умолчанию
        // Мониторим через JMX:
        // kafka.consumer:type=consumer-fetch-manager-metrics
        // kafka.consumer:type=consumer-coordinator-metrics
        
        KafkaConsumer<String, String> consumer = 
            new KafkaConsumer<>(props);
        
        // Запрос метрик
        consumer.metrics().forEach((metricName, metric) -> {
            System.out.println(metricName + ": " + metric.value());
        });
    }
}

Best Practices для репликации

1. Для критичных данных: используйте acks=all с min.insync.replicas=2
2. Мониторьте ISR: убедитесь, что реплики синхронизированы
3. Устанавливайте правильный replication factor: минимум 3 для production
4. Балансируйте нагрузку: распределяйте лидеров между брокерами
5. Тестируйте failover: убедитесь в восстановлении после падений

Когда использовать разные уровни репликации

• acks=0: логирование, метрики, non-critical данные
• acks=1: большинство бизнес-случаев
• acks=all: финансовые операции, критичные данные, гарантия доставки

Репликация в Kafka — это фундаментальный механизм, обеспечивающий надёжность и отказоустойчивость в распределённых системах обработки потоков данных.