📨 Розділ 15 · Питання #29

Чи можна читати повідомлення з певної партиції

В Apache Kafka консьюмер може читати повідомлення безпосередньо з конкретно вказаних партицій топіка. Для цього замість механізму динамічної підписки consumer.subscribe() викори...

📨 Kafka 📄 Вихідний Markdown

🟢 Junior Level

Пряма відповідь: Так, можна!

В Apache Kafka консьюмер може читати повідомлення безпосередньо з конкретно вказаних партицій топіка. Для цього замість механізму динамічної підписки consumer.subscribe() використовується метод явного ручного призначення — consumer.assign().

Динамічна підписка (subscribe):
Consumer ──subscribe("orders")──> Group Coordinator на брокері сам вирішує,
                                  які партиції призначити (наприклад, партиції 0 і 1).

Ручне призначення (assign):
Consumer ──assign([orders-0])───> Читає СТРОГО і ТІЛЬКИ партицію 0.
                                  Брокер не бере участі в розподілі.

Базовий приклад на Java

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 1. Формуємо список цільових партицій (наприклад, партиція 0 топіка "orders")
TopicPartition partition0 = new TopicPartition("orders", 0);
consumer.assign(List.of(partition0));

// 2. Читаємо повідомлення в стандартному циклі poll
while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
            record.partition(), record.offset(), record.key(), record.value());
    }
}

Призначення в Spring Kafka

У фреймворку Spring Kafka пряме читання конкретних партицій налаштовується декларативно за допомогою анотації @KafkaListener:

@KafkaListener(
    topicPartitions = @org.springframework.kafka.annotation.TopicPartition(
        topic = "orders", 
        partitions = { "0", "2" } // Слухаємо тільки партиції 0 та 2
    )
)
public void listenSpecificPartitions(ConsumerRecord<String, String> record) {
    log.info("Отримано повідомлення з партиції {}: {}", record.partition(), record.value());
}

🟡 Middle Level

Фундаментальні відмінності: subscribe() проти assign()

Методи subscribe() та assign() є взаємовиключними (mutually exclusive). Спроба викликати обидва методи на одному екземплярі KafkaConsumer призведе до викидання винятку IllegalStateException.

Характеристика consumer.subscribe() consumer.assign()
Координатор групи Активно використовується (GroupCoordinator) Повністю ігнорується
Ребалансування (Rebalance) Автоматичне при падінні або масштабуванні Відсутнє. Консьюмер автономний (Standalone)
Відмовостійкість Висока: партиція впалого пода відійде сусіду Нульова: при аварії партиція перестане вичитуватися
Наявність group.id Обов’язкова Опціональна (потрібна лише для фіксації зміщень)
Додавання нових партицій Виявляються автоматично Ігноруються (потрібен ручний перевиклику assign())
Сфера застосування Стандартна продакшн-обробка мікросервісів Реплей, налагодження, аудит, StatefulSet шардування

Керування зміщенням (Seek API)

Головна перевага assign() розкривається у поєднанні з довільним позиціонуванням читання:

TopicPartition tp = new TopicPartition("orders", 0);
consumer.assign(List.of(tp));

// 1. Читання з самого початку партиції (з найменшого доступного LogStartOffset)
consumer.seekToBeginning(List.of(tp));

// 2. Читання тільки нових повідомлень (з кінця логу LogEndOffset)
consumer.seekToEnd(List.of(tp));

// 3. Читання з точного номера зміщення
consumer.seek(tp, 450200L);

// 4. Пошук зміщення за часовою міткою (Time-based Seek)
long oneHourAgo = System.currentTimeMillis() - TimeUnit.HOURS.toMillis(1);
Map<TopicPartition, Long> timestampsToSearch = Map.of(tp, oneHourAgo);

// Запитуємо у брокера перше зміщення, записане після вказаного часу
Map<TopicPartition, OffsetAndTimestamp> foundOffsets = consumer.offsetsForTimes(timestampsToSearch);
OffsetAndTimestamp target = foundOffsets.get(tp);

if (target != null) {
    consumer.seek(tp, target.offset());
}

🔴 Senior Level

Робота з group.id та __consumer_offsets при ручному призначенні

Поширена помилкова думка: «При використанні assign() параметр group.id не потрібен і комітити офсети не можна».

Це не так. Механіка роботи залежить від конфігурації:

  1. group.id не вказано:
    • Виклики consumer.commitSync() або consumer.commitAsync() викинуть InvalidGroupIdException: To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.
    • Консьюмер не зберігає стан на брокері. При кожному перезапуску він читатиме з позиції, визначеної auto.offset.reset (якщо не було явного виклику seek()).
  2. group.id вказано:
    • Консьюмер НЕ бере участі в протоколі балансування групи (не відправляє запити JoinGroup / SyncGroup).
    • Проте при виклику commitSync() координатор брокера зберігає зміщення в системному топіку __consumer_offsets під зазначеним group.id!
    • При перезапуску застосунку consumer.assign() автоматично прочитає останній закомічений офсет із __consumer_offsets і продовжить обробку без повторів.

Production-архітектура: Патерн StatefulSet + Partition Pinning

У деяких сценаріях динамічне ребалансування subscribe() є неприйнятним (Stop-the-World паузи при ребалансі, скидання локальних кешів). У розподілених базах даних та аналітичних рушіях (Apache Flink, Kafka Streams, локальні кеші на RocksDB) використовується патерн Deterministic Partition Pinning:

Kubernetes StatefulSet (order-worker):
Pod 0 (HOSTNAME="order-worker-0") ──> assign(TopicPartition("orders", 0)) ──> Локальний RocksDB 0
Pod 1 (HOSTNAME="order-worker-1") ──> assign(TopicPartition("orders", 1)) ──> Локальний RocksDB 1
Pod 2 (HOSTNAME="order-worker-2") ──> assign(TopicPartition("orders", 2)) ──> Локальний RocksDB 2

Переваги:

  1. Повна відсутність пауз на ребалансування при деплої чи перезапуску подів.
  2. Локальний дисковий стан (Embedded RocksDB) жорстко закріплений за партицією і не перекачується мережею між подами при рестартах.
  3. Простота розслідування інцидентів: логування партиції ізольоване в конкретному контейнері.

Недоліки:

Якщо Pod 1 падає через OOM, Kubernetes зобов’язаний перезапустити саме його. Доки под піднімається, дані партиції 1 не обробляються ніким, оскільки інші поди не можуть автоматично перехопити чужу партицію.

4 Tricky Questions

1. Що станеться, якщо у працюючому застосунку спочатку викликати consumer.subscribe(List.of("topicA")), а потім на тому самому об’єкті викликати consumer.assign(List.of(new TopicPartition("topicB", 0)))?

Відповідь: Виклик завершиться винятком IllegalStateException: java.lang.IllegalStateException: Subscription to topics, partitions and pattern are mutually exclusive. Усередині KafkaConsumer поле SubscriptionState зберігає тип поточного призначення: NONE, AUTO_TOPICS, AUTO_PATTERN або USER_ASSIGNED. Перемикання між динамічною підпискою та ручним призначенням можливе тільки після явного виклику consumer.unsubscribe(), який очищає поточний стан і скидає з’єднання з координатором групи.

2. Адміністратор збільшив кількість партицій у топіку з 4 до 8 на працюючому кластері. Як відреагують консьюмери зі subscribe() та консьюмери з assign()?

Відповідь:

  1. Консьюмери зі subscribe(): Фоновий потік оновлення метаданих регулярно опитує кластер (metadata.max.age.ms, за замовчуванням 5 хвилин). Виявивши нові партиції, консьюмери ініціюють ребалансування групи, і нові партиції 4..7 будуть автоматично розподілені між учасниками групи.
  2. Консьюмери з assign(): Вони не побачать нових партицій ніколи. Список призначених партицій зафіксований у пам’яті клієнта статично. Застосунок продовжуватиме вичитувати лише початкові партиції 0..3. Щоб почати читати партиції 4..7, сервіс зобов’язаний або перезапуститися, або періодично програмно опитувати consumer.partitionsFor("topic") та вручну викликати consumer.assign(updatedPartitions).

3. Чи працюватиме ConsumerRebalanceListener при використанні consumer.assign()?

Відповідь: Ні, ніколи. Інтерфейс ConsumerRebalanceListener із методами onPartitionsRevoked() та onPartitionsAssigned() передається лише як аргумент методу subscribe(). Метод assign() приймає виключно колекцію TopicPartition і не має перевантажень для слухачів. Оскільки протокол координації групи не задіюється, подій відкликання або призначення партицій з боку брокера взагалі не генерується.

4. Чи можна організувати Exactly-Once обробку із записом у реляційну БД, використовуючи assign() замість subscribe()?

Відповідь: Так, і це один із найбільш надійних класичних патернів.

  1. Консьюмер закріплює за собою партицію через assign(tp).
  2. Застосунок читає повідомлення та записує бізнес-дані в PostgreSQL або Oracle.
  3. В ту саму транзакцію бази даних (у службову таблицю kafka_offsets) зберігається пара (partition_id, processed_offset): 
    BEGIN TRANSACTION;
INSERT INTO orders (id, amount) VALUES (1, 100);
UPDATE kafka_offsets SET offset = 105 WHERE partition = 0;
COMMIT;
  4. При старті або після збою сервіс виконує: long lastOffset = db.query("SELECT offset FROM kafka_offsets WHERE partition = 0"); consumer.seek(tp, lastOffset + 1);
  5. Оскільки зміщення фіксується атомарно разом із бізнес-даними в єдиній транзакції БД, дублікати та втрати даних повністю виключені (сувора семантика Exactly-Once без необхідності використання важкого Kafka 2PC Transactions API).

🎯 Шпаргалка для інтерв’ю

30-секундна відповідь (Elevator Pitch)

«Так, читати конкретну партицію можна за допомогою методу consumer.assign(Collection<TopicPartition>). На відміну від subscribe(), ручне призначення повністю вимикає протокол координації групи та ребалансування — консьюмер функціонує як цілком автономний процес (Standalone Consumer). Це забезпечує абсолютний детермінований контроль над партиціями та дозволяє використовувати довільне позиціонування через Seek API (seek(), seekToBeginning(), offsetsForTimes()). Плата за це — відсутність вбудованої відмовостійкості: якщо консьюмер з assign() падає, його партиції автоматично нікому не передаються. Основні застосування: реплей даних, налагодження, аудит, а також шардовані архітектури з локальним станом (StatefulSets + RocksDB)».

Ключові метрики для Production-моніторингу

  • kafka.consumer:type=consumer-fetch-manager-metrics,client-id={id} -> records-lag: лаг конкретної призначеної партиції.
  • Статус процесу подів у Kubernetes: оскільки автоматичного перехоплення партицій немає, падіння пода вимагає негайного алертингу.

Червоні прапорці (Чого говорити НЕ МОЖНА)

  • ❌ «subscribe() та assign() можна викликати одночасно на одному консьюмері» — Це призведе до негайного винятку IllegalStateException.
  • ❌ «При assign() не можна комітити офсети в Kafka» — Можна, якщо вказано group.id; зміщення будуть надійно зберігатися в __consumer_offsets.
  • ❌ «Якщо под з assign() упаде, інший под автоматично підхопить його партицію» — Ні, при ручному призначенні координатор групи вимкнений, обробка партиції зупиниться до підняття пода.
  • ❌ «Нові партиції, створені адміністратором на льоту, автоматично з’являться в assign()» — Ні, список статичний у пам’яті клієнта, потрібен явний перевиклику assign().

Пов’язані теми