Стримінг має проблему тиші.

Коротка версія

Коли партиція Kafka замовкає — вона завершилася чи просто повільна? Ззовні це виглядає однаково — а відповідь визначає, чи можна безпечно діяти на основі того, що у вас є. Netflix, Uber і Walmart опублікували свої підходи, і кожен чимось жертвує: другим конвеєром, смугою паралелізму, коректністю під час повторного відтворення. Щоразу — добротна інженерія. Але жоден не питає у Kafka того єдиного, що Kafka вже знає.

Я зіткнувся з цим на потоці інвентаризації для компанії з Fortune 50: тринадцять топіків Kafka, по три партиції в кожному, сотні мільйонів подій на день, кожна подія для певного товару мала йти в строгому порядку за часом події — читання семи днів історії з нуля з негайним переходом до пікового живого трафіку. Перший дизайн був коректним, але повільним: тридцятихвилинне вікно лише на зчитування історії, сплачуване знову при кожному перезапуску, плюс друге завдання, що підбирало живий трафік. Уся задача полягала в тому, щоб відіграти цей час назад, не дозволивши жодному запису випасти з порядку.

Версія, що пішла у продакшн, не чекає на годинник. Вона читає офсети Kafka.

Ось як — і чому ця ідея старша за Kafka.

Компроміс, на який ідуть усі / §01

Це не проблема однієї команди. Чотири сильні інженерні організації, чотири обґрунтовані компроміси:

• Walmart закріплює кожну пару «товар–магазин» за єдиною партицією, тож її завжди оновлює лише один споживач. Порядок через закріплення ключів — ціною паралелізму.
• Uber запускає окремий конвеєр дозаповнення для переобчислення після запізнілих і невпорядкованих подій, бо потоковий шлях «може спричиняти неточності, відкидаючи будь-які події, що надходять після встановлення вотермарки». Ціна — другий конвеєр.
• Netflix, дозаповнюючи конвеєри Flink, виявив, що відтворена історія дає інший порядок і некерований стан. Ціна — коректність і стан.
• Confluent у власних рекомендаціях щодо простоюючої партиції Kafka радить: чекати на неї і зупинятися — або ігнорувати її і приймати «багато запізнілих повідомлень, щойно партиція знову стане активною». Ціна — зупинки або запізнілі дані.

Кожен варіант розумний. Спільна нитка в тому, що всі вони виводять повноту з часу — вікно, тайм-аут, очікування — тоді як офсети партицій точно кажуть, де закінчуються дані. Саме цю прогалину й закриває це рішення.

Чому потік переплутаний / §02

Топік Kafka розбито на партиції, і кожна партиція — це власна колія. Усередині партиції порядок гарантовано: запис перший, потім запис одинадцятий, завжди. Між партиціями ніщо нічого не координує — споживачі, що їх читають, ніколи не перебувають в одному місці водночас. Kafka робить рівно те, що обіцяє, споживачі роблять рівно те, що обіцяють, а об'єднаний вихід усе одно переплутаний для всього, що залежить від послідовності.

Не можна діяти на «продано» раніше за «в наявності» для однієї і тієї ж одиниці. Тож перш ніж конвеєр щось випускає, він мусить відповісти на одне запитання: чи отримав я все до цього моменту? Майже кожна відповідь, що використовується сьогодні, — це таймер: вікно, тайм-аут простою, завдання звірки. Кожна з них — здогадка про тишу.

Старе рішення: повідомляй, а не чекай / §03

Є прецедент, і це не кращий годинник. Ранні залізниці працювали за часовими інтервалами: наступний потяг вирушав, щойно минав фіксований інтервал, отже колія «мусила» бути вільною.

У червні 1889 року потяг застряг на крутому підйомі під Армою, в Ірландії. Задні вагони відчепили, вони покотилися назад з гори і врізалися в потяг, що йшов слідом. Загинуло близько вісімдесяти людей.

Наступний потяг вирушив з Арми, щойно минули обов'язкові двадцять хвилин. Двадцять хвилин минуло, отже колія мала бути вільною. Ніхто в Армі не мав жодного способу дізнатися, що це не так.

Британія не відповіла кращим годинником. Парламент надав Торговельній раді повноваження вимагати на пасажирських залізницях абсолютну блокову систему: жоден потяг не входить на перегін, доки сигнальний пост попереду не підтвердить, що перегін вільний. Вільність перестала бути тривалістю, яку хтось вичікував, і стала фактом, про який хтось повідомляв. Застряглий потяг і потяг, що відбув, більше не виглядали однаково, бо ніхто вже не вичитував нічого з тиші.

Це і є вся суть. Читання офсетів Kafka — це сигнальний пост.

Побудова — три складові / §04

Принцип короткий: час рухається лише тоді, коли докази підтверджують, що він може. Щоб утілити це, знадобилися три складові, і найважча частина була не там, де очікують.

Перша: черга таймерів — це і є сортування. Кожен запис паркується у ключованому стані під власним часом події, з таймером часу події на цю точну мілісекунду. Поки що нічого не випускається. Коли вотермарка досягає цієї позначки часу, таймер спрацьовує, і все припарковане там виходить разом. Flink спрацьовує таймери ключа у порядку зростання позначок часу, тож ми ніколи нічого не сортуємо. Це один метод і шматок стану — складність ніколи не була тут.

Друга: потік без часу події стає годинником. Один із тринадцяти топіків узагалі не мав часу події — він лише публікував дійсні товари, якими збагачувалося все інше. Отже, йому не було чого сортувати, і він міг робити те, чого не могли інші: випускати вотермарки з часу обробки, завжди найпізнішу можливу позначку. Але є нюанс: годинник конвеєра — це мінімум серед його входів, тож потік, що біжить на «зараз», нічого не змінює, доки історичні потоки ще дочитуються. Я побудував цю складову першою, і вона не дала жодного ефекту — вона не була помилковою, вона була не в тому порядку. Це був прорив: щойно інші потоки стають простоюючими, керуючий потік перебирає керування, годинник стрибає до «зараз», хвіст виливається одним сплеском, і ніщо не зупиняється, чекаючи на тихий топік. Проте це важить лише тоді, коли інші виходять із мінімуму — а потік виходить із мінімуму, коли його позначено простоюючим.

Третя: коли потік завершено, вирішує Kafka — а не тайм-аут. Вбудований механізм простою у Flink — це таймер: немає даних якийсь час — позначити простоюючим. Але тайм-аут не відрізнить «запускається» від «завершився», а запуск — найгірший випадок: з'єднання встановлюються з різною швидкістю, один топік сидить тихо, тайм-аут спрацьовує, інші проганяють годинник повз часи подій, яких він ще не прочитав, і його перший запис прибуває вже запізнілим. Це не затримка, це порушення. Тож тайм-аут мусив піти. При запуску, ще до читання будь-якого запису, завдання запитує в Kafka останній офсет у кожній партиції і тримає його. Топік простоює лише тоді, коли кожна партиція досягла цього числа — не більшість, кожна. Жодної тривалості для налаштування, бо вентилем служить факт від Kafka.

// once, at job startup
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(partitions);

// as records arrive
if (offset + 1 >= startupEndOffsets.get(partition)) markCaughtUp(partition);

Що це дало / §05

Вікно утримувало кожен запис до свого закриття — тридцять хвилин лише на зчитування історії, і знову при кожному перезапуску. Тепер запис випускається, щойно вотермарка його проходить: у середньому менш ніж 100 мілісекунд, включно із записами, притриманими, щоб вони не вийшли не по черзі. Запис, що прибуває на першій хвилині, виходить на першій хвилині. З порядком — так само: два шари простою за тайм-аутом випускали 5,5% записів не по черзі; один шар — 1,5%; простій, що його визначають стартові офсети Kafka, — нуль. Вікно зникло, і завдання-підбирач теж — одне завдання читає сім днів історії і далі працює наживо.

Чого це не обіцяє / §06

Детермінованість — не ясновидіння. Подія існує для конвеєра лише з моменту, коли вона потрапляє в Kafka. Якщо продюсер притримує запис і той потрапляє туди після того, як годинник уже пройшов його час події, він виходить негайно, поза чергою, і лічильник порушень це фіксує. Жоден потоковий процесор не може впорядкувати подію, якої він ще не бачить. Усередині цієї межі гарантія точна, а ширину межі задає відставання вотермарки.

Те саме запитання — всюди / §07

Побачивши це раз, ви бачитимете це в кожній системі, яка визначає істину з потоків, що не надходять одночасно.

Річковий водомірний пост замовкає під час повені. Річка стабільна — чи вода дісталася до сенсора? У 2018 році USGS повідомила про гідрометричні пости, виведені з ладу ураганом «Флоренс» — саме ті, що мали найбільше значення. «Немає новин» — хибне прочитання такої тиші.

Система камер. Знімаєте спортсмена кількома камерами — і кожна є власним потоком, як партиція; ви не можете зібрати 3D-позу для миті, доки кожна камера її не повідомила. Мовчазна камера могла втратити кадр — або спостерігати за кимось, хто завмер. На i3d.ai це те саме запитання, десятки разів на секунду.

Агент. AI-агент викликає інструмент і не чує нічого у відповідь. Агенти дуже добре вміють трактувати тишу як успіх — помилки не прийшло, отже далі, до наступного кроку з машинною швидкістю. Агент, що діє на обліковому записі клієнта в стані годинної давності, — це потяг, відправлений за таймером. Виправлення те саме, що знайшли залізниці: жоден крок не відбувається на тиші; кожен крок відбувається на підтвердженні, на яке можна вказати.

Найдешевший виграш у надійності стримінгу — не швидший рушій і не більше вікно. Це перестати виводити «завершено» з тиші й запитати систему, яка вже знає. Вікно вичікує найгірший випадок для кожного запису; перевірка офсетів чекає рівно стільки, скільки вимагають докази, і ні на мілісекунду довше.

Спрощена еталонна реалізація — генератор, обробник, трекер наздоганяння — це наступне, що я опублікую.