Безошибочная работа с Kafka из Node js. Часть 3 Cтруктура сообщений, когда Kafka не нужна и теряет данные

await producer.send({
  topic: '...',
  messages: [
    {
      key: '...',
      value: JSON.stringify({
        message_id: randomUUID(),
        timestamp: date,
        sequence_id: number,
        payload_v1: {
          // data...
        }
      }),
      headers: {
        'x-service-name': string,
        'x-request-id': string
      }
    }
  ]
});

• x-service-name – имя сервиса, который опубликовал сообщение. Полезно для трассировки. Например, когда в один топик пишут несколько сервисов или сообщение может пройти путь через несколько топиков прежде, чем достигнуть консьюмера.
  
• x-request-id – уникальный id запроса. Необходим для удобного логирования, например, отфильтровать логи, cвязанные c пачкой сообщений или одним конкретным.
  
• message_id – уникальный индентификатор сообщения. Позволяет как определить дубли записей, так и защититься от повторной обработки на стороне консьюмера.
  
• timestamp или sequence_id – поля, определяющие порядок создания сообщения на продьюсере. Kafka дает гарантии порядка для уже записанных в топик сообщений, но иногда они могут быть записаны в брокер в разном порядке. Например, пользователь дважды обновил свою фотографию в приложении в рамках малого промежутка времени. Первое обновление заняло больше времени (фото долго грузилось в S3), чем второе, как результат в топик записалось сначала второе событие, а потом первое. Пользователь увидит в приложении предпоследнее фото вместо последней загруженной фотографии. Для решения данной проблемы требуется передавать ключ порядка операций:
  • timestamp – время создания сообщения на продьюсере, а не время, когда его отправили в топик (producer.send)
    
  • sequence_id – уникальный монотонно возрастающий идентификатор. Можно использовать sequence_id, когда timestamp не подходит, например, когда не хочется полагаться на синхронизацию часов у всех инстансов приложения. По аналогии с timestamp приложение должно запрашивать sequence_id в самом начале бизнес операции, а не в момент отправки сообщения в топик (producer.send). Примером sequence_id, может служить autoincrement поле или функция nextval в БД.
• payload_v* – передаваемые данные. Для того чтобы иметь возможность версионировать сообщения в ситуации, когда невозможно сохранить обратную совместимость структуры, используется постфикс версии. Почему версионирование нельзя реализовывать через создание нового топика, например, topic_v2? Версионирование через топик не позволяет консьюмеру обеспечить порядок обработки сообщений, ведь сообщения разных версий попадут в разные партиции и топики. Кроме того, в случае отката на предыдущую версию возникают проблемы с необработанными сообщениями в topic_v2: их надо переместить в топик v1 (Kafka не имеет встроенного инструмента в отличие от RabbitMQ) и трансформировать к предыдущей версии. В случае версионирования тела сообщения один и тот же консьюмер сможет по условию или фичи флагу обрабатывать разные версии сообщения c гарантией порядка обработки. А при переключении версии не будет возникать проблем с перезалитием сообщений из одного топика в другой.

• Подключение 1000+ клиентов. Кластер Kafka плохо поддерживает подключение тысячи и более клиентов. Например, очень плохая идея подключать тысячи мобильных устройств в качестве продьюсеров или консьюмеров. Рекомендуется поставить proxy/gateway перед Kafka, к которому по HTTP или MQTT будут подключаться устройства для получения или публикации данных.
  
• Плохая сеть. Kafka чувствительна к стабильности сетевого соединения между клиентами и брокерами. В условиях плохой сети у консьюмеров будут частые ребалансы, продьюсеры в процессе отправки могут терять сообщения, а брокер может терять данные или быть недоступен для записи. В сочетании с указанным выше пунктом это может стать cмертью для кластера. В условиях плохой сети рекомендуется использовать другие технологии: HTTP, MQTT, UDP и т.д.
  
• Реалтайм. Kafka не подходит для систем, где требуется передача данных строго в реальном времени. Очевидно, что сама Kafka, как промежуточное звено добавляет лаг задержки. И к тому же в системах реального времени отдается предпочтение скорости доставки, чем надежности и упорядоченности событий. Например, в трейдинге или игровом шутере требуется получать текущее значение котировки/противников как можно быстрее и чаще, нежели иметь возможность упорядоченно обработать события или заново их перечитать. Для функионала, где требуется реалтайм работа стоит использовать: HTTP/2-3 (SSE или QUICK), WebSocket, HLS, WebRTC, RTMP, MQTT.
  
• Репликация данных. Для периодической синхронизации данных между несколькими БД гораздо лучше подойдут снятие и применения дампов, чем передача данных через Kafka. Для перманентной синхронизации с минимальной задержкой следует использовать нативную репликацию самой БД. Обычно это отлаженный механизм, на который можно положиться и который не нуждается в переизобретении. Для синхронизации разных БД (MySQL –> MongoDB) предпочтительней использовать готовые ETL движки, полагающиеся на журнал БД: binlog, wal, oplog и т.д.
  
• RPC или request/reply-to. Kafka не подходит в качестве транспорта для RPC взаимодейстия. Ее не проектировали для создания топиков и партиций «на лету» в отличие от того же RabbitMQ. При этом большое их количество отрицательно влияет на производительность всего кластера, ведь это создает нагрузку на ZooKeeper/KRaft. Для RPC взаимодействия следует отдать предпочтение другим инструментам: HTTP webhook, RPC (JSON-RPC, SOAP, gRPC), RabbitMQ, NATS.
  • База данных. ksqlDB дает возможность рассматривать Kafka в качестве БД. Данное решение под капотом использует Kafka Streamsи RocksDB в качестве хранилища данных. Но использование ksqlDB имеет ряд проблем.
    • Отсутствие индексов (вторичных), ведь RocksDB это key-value БД.
      
    • Сложности с join. ksqlDB стриминговая БД поэтому есть необходимость создавать window (сообщения должны попадать в определенной временной диапазон, например, последние 10 минут). Данные для join должны находиться в одной партиции, иначе их придется перезаливать в новый топик, что создает дополнительную сетевую нагрузку и увеличивает объем хранимых данных на диске.
      
    • При изменении запросов нужно перепрочитать все данные в топике. При создании нового запроса (CREATE TABLE AS SELECT) или редактировании уже существующего, ksqlDB прочитает весь топик с самого начала. Выполнение запроса на больших топиках может занимать много времени, создавать нагрузку на кластер и потреблять много RAM (ksqlDB читает сообщение целиком, а не отдельные поля).
      
    • Использование ksqlDBсильно увеличивает объем потребляемого пространства на диске. Во-первых, потребители данных в БД c большой вероятностью попросят увеличить retentionдля топиков. Для того чтобы иметь возможность посчитать актуальные данные для больших window в недавно созданных запросах и иметь возможность перепрочитать сообщения за длительный период. Во-вторых, ksqlDB имеет проблему дублирования при хранении данных. Это происходит из-за таких сущностей как stream, table, changelog topic на которых построена вся обработка данных в ksqlDB.
    Как итог, Kafka в качестве БД является плохой идей. Cтоит использовать реляционные БД такие MySQL/PostgreSQL (индексы и обычные join) или колоночные/timeseries БД (сжатие данных на диске и быстрый доступ к ним), например, ClickHouse/QuestDB.
• Cложная топология потребителей. Существуют архитектурные сценарии, когда конcьюмерам надо читать лишь определенные типы сообщений. Например, консьюмер хочет получать все события платежей во всех магазинах Москва: moscow.*.payment.* (схема регион.id_магазина.функционал.имя_события). Cложную топологию в рамках Kafka можно организовать следующими способами:
  • Создание множества специализированных топиков под требования консьюмеров. Kafka позволяет консьюмерам подписаться на несколько топиков с помощью регулярного выражение. Однако такой подход создает нагрузку на кластер. ZooKeeper/KRaft испытывает проблемы с тем чтобы управлять тысячами топиков и партиций.
    
  • Использовать один топик для всех типов сообщений, а консьюмеры отфильтровывают целевые события. Недостаток такого подхода в том, что консьюмерам надо пропускать через себя большое количество ненужных данных.
  Исходя из выше сказанного, для сложной топологии вместо Kafka cтоит выбирать RabbitMQ или Apache Pulsar.
  
• Передача больших данных. Kafka испытвает проблемы при передаче больших сообщений (> 1 Mb и тем более > 10Mb). Подробно об этом говорилось в предыдущей части. Данные следует дробить на части или использовать внешние системы для хранения, а в сообщение лишь передавать ссылку на них.
  
• DLQ (Dead Letter Queue) и очереди с приоритетами.Подход DLQ можно реализовать в Kafka, но только в случае, когда не важен порядок обработки. В противном случае сообщения из основного топика могут быть обработаны раньше, чем сообщения, которые переместили из него в DLQ. Очереди c приоритетом можно организовать в Kafka, но это довольно трудозатратно и неудобно. И для DLQ и для очередей с приоритетом гораздо правильнее использовать RabbitMQ.

• Продьюсер не успел отправить накопленный буффер сообщений по причине перезапуска его или ошибки.
  
• Продьюсер не успел вызвать метод .send по аналогичным причинам.
  
• Продьюсер не смог отправить сообщения (с учетом retry) в брокер из-за проблем с сетью или недоступности лидер ноды партиции.

• Часть 1. Продьюсер
  
• Часть 2. Консьюмер