Налаштування топіків та партицій Kafka

Налаштування топіків та партицій Kafka

Кількість партицій та коефіцієнт репліцирання — два параметри, які не можна легко змінити після створення топіку. Зменшити кількість партицій неможливо без повного перестворення топіку. Тому правильне налаштування при створенні важливо.

Як працюють партиції

Партиція — це одиниця паралелізму в Kafka. Один консьюмер у групі обробляє одну партицію. Якщо топік має 6 партицій, максимум 6 консьюмерів у групі можуть читати паралельно. Зайві консьюмери простоюють.

Запис у партицію суворо впорядкований. Глобальний порядок по топіку не гарантується — тільки в межах партиції. Це важливо для подій, які мають обробляються послідовно (всі дії одного користувача).

Topic: user-events (6 партицій)
Partition 0: event1(user:101), event4(user:205), ...
Partition 1: event2(user:102), event5(user:101), ...  ← user:101 розбиті по партиціях!
Partition 2: event3(user:103), ...

Щоб забезпечити, що дії одного користувача йдуть в одну партицію — використовуємо ключ повідомлення:

Запис з ключем user_id → hash(user_id) % num_partitions → завжди одна партиція

Розрахунок кількості партицій

Практичне правило: num_partitions = max(throughput_target / throughput_per_partition, num_consumers_target).

Типова пропускна здатність однієї партиції: 10–50 MB/s для запису (залежить від залізо та конфігурації брокера).

Приклад: потрібно обробляти 200 MB/s з піком до 400 MB/s та мати можливість масштабування до 20 консьюмерів → беремо 24 партиції (кратно 6, 8, 12 для зручного масштабування).

Занадто багато партицій — теж погано: кожна партиція потребує filehandle, пам'ять для буферів, перевантажує контролер при виборах лідера.

Створення топіків через kafka-topics.sh

# Базовий топік для подій користувачів
kafka-topics.sh --bootstrap-server kafka-1:9092 \
    --create \
    --topic user-events \
    --partitions 12 \
    --replication-factor 3 \
    --config retention.ms=604800000 \
    --config retention.bytes=10737418240 \
    --config compression.type=lz4 \
    --config min.insync.replicas=2 \
    --config message.max.bytes=1048576

# Компактний топік — для зберігання останнього стану за ключем
kafka-topics.sh --bootstrap-server kafka-1:9092 \
    --create \
    --topic user-profiles \
    --partitions 24 \
    --replication-factor 3 \
    --config cleanup.policy=compact \
    --config min.cleanable.dirty.ratio=0.1 \
    --config segment.ms=3600000 \
    --config delete.retention.ms=86400000

# Високопріоритетна черга з коротким утриманням
kafka-topics.sh --bootstrap-server kafka-1:9092 \
    --create \
    --topic order-processing-priority \
    --partitions 6 \
    --replication-factor 3 \
    --config retention.ms=3600000 \
    --config max.message.bytes=102400

Управління через Admin API (Java/Kotlin)

Створення топіків програмно — правильно для застосунків, які створюють топіки динамічно:

Properties props = new Properties();
props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-1:9092,kafka-2:9092,kafka-3:9092");
props.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 5000);
props.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 10000);

try (AdminClient admin = AdminClient.create(props)) {
    NewTopic userEvents = new NewTopic("user-events", 12, (short) 3);
    userEvents.configs(Map.of(
        "retention.ms", "604800000",
        "compression.type", "lz4",
        "min.insync.replicas", "2"
    ));

    NewTopic deadLetter = new NewTopic("user-events-dlq", 3, (short) 3);
    deadLetter.configs(Map.of(
        "retention.ms", "2592000000", // 30 днів
        "retention.bytes", "-1"
    ));

    CreateTopicsResult result = admin.createTopics(List.of(userEvents, deadLetter));
    result.all().get(30, TimeUnit.SECONDS);
}

Зміна конфігурації існуючого топіку

# Збільшуємо утримання
kafka-configs.sh --bootstrap-server kafka-1:9092 \
    --alter \
    --entity-type topics \
    --entity-name user-events \
    --add-config retention.ms=1209600000

# Додаємо партиції (тільки збільшення!)
kafka-topics.sh --bootstrap-server kafka-1:9092 \
    --alter \
    --topic user-events \
    --partitions 24

# Увага: додавання партицій порушує порядок для ключевих повідомлень.
# Існуючі ключі йдуть у ті ж партиції (hash % 12),
# нові ключі розподілятимуться по 24 партиціях.

# Перегляд конфігурації топіку
kafka-configs.sh --bootstrap-server kafka-1:9092 \
    --describe \
    --entity-type topics \
    --entity-name user-events

Управління лідерами партицій

Нерівномірне розподіл лідерів між брокерами призводить до гарячих вузлів:

# Перевіряємо розподіл лідерів
kafka-topics.sh --bootstrap-server kafka-1:9092 \
    --describe --topic user-events

# Перебалансування за кількістю реплік
kafka-leader-election.sh --bootstrap-server kafka-1:9092 \
    --election-type preferred \
    --all-topic-partitions

# Або для конкретних партицій через JSON
cat > election.json << 'EOF'
{
  "partitions": [
    {"topic": "user-events", "partition": 0},
    {"topic": "user-events", "partition": 1}
  ]
}
EOF

kafka-leader-election.sh --bootstrap-server kafka-1:9092 \
    --election-type preferred \
    --path-to-json-file election.json

Моніторинг партицій

# Consumer lag — відставання групи
kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
    --describe --group my-consumer-group

# Вивід: TOPIC / PARTITION / CURRENT-OFFSET / LOG-END-OFFSET / LAG
# Сумарний lag > 10000 для критичних топіків — повід для алерту

# Скидання offset (якщо потрібно перечитати з початку)
kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 \
    --group my-consumer-group \
    --topic user-events \
    --reset-offsets --to-earliest \
    --execute

Типові конфігурації за типом даних

Таймлайн

День 1 — аналіз вимог: пропускна здатність, кількість консьюмерів, вимоги до впорядкованості, утримання. Проектування схеми топіків.

День 2 — створення топіків, налаштування ACL (якщо увімкнена аутентифікація), тестування producer/consumer з правильними ключами.

День 3 — налаштування моніторингу consumer lag, алерти, документування схеми для команди.