1С KAFKA. НЕ ПИШИТЕ СВОЙ REST СЕРВИС ДЛЯ КАФКИ

1С KAFKA. НЕ ПИШИТЕ СВОЙ REST СЕРВИС ДЛЯ КАФКИ НА ПИТОНЕ.

В телеграмм чатах, на профильных конференциях я часто сталкиваюсь с кейсом когда 1С программисты в проектах внедрения «Kafka для 1С» принимают неоптимальное решение. Обычно оно звучит так – «МЫ написали свой REST сервис». Часто в качестве платформы для написания выбирается Django. Кстати – такую ошибку делают не только 1С-ники, но и соседи (java, Python, .NET, etc разработчики).

Почему данное решение НЕ оптимально? Помимо того, что Django для написания простого Web сервиса – это примерно, как забивать гвозди промышленным плиткорезом, что конечно можно, но не профильно. Так вот, помимо этого, если Вам прилетела задача интегрироваться с Apache Kafka из 1С, а денег на покупку штатной librdkafka не выделили, есть специализированный сервис от компании Confluent – Kafka Rest Proxy.

На нашем вебинаре по Apache Kafka и 1С мы разбирали этот момент подробно, однако в рамках данной статьи попробуем разобраться, как использовать штатный REST Proxy для Apache Kafka из 1С — с рабочими примерами кода.

Когда Kafka REST Proxy, когда HTTP-сервис 1С, а когда — 1С:Шина

Прежде чем писать код, стоит определиться, какой инструмент вообще нужен под задачу. Это разные классы решений, и путаница между ними — источник половины проблем в интеграционных проектах 1С.

Коротко: если 1С нужно просто отправлять и получать события из Kafka — берите REST Proxy и не пишите свой сервис, в этом главная мысль статьи. Если задача — синхронный запрос-ответ без брокера сообщений, для этого достаточно обычного HTTP-сервиса 1С. А если систем много и нужна оркестрация потоков между ними — смотрите в сторону 1С:Шина: она умеет работать поверх Kafka и RabbitMQ как ESB-прослойка, а не заменяет их в лоб.

Быстрый старт: поднимаем Kafka REST Proxy локально

Маленькое дополнение – все примеры кода будут даны для использования с подсистемой Connector от Владимира Бондаревского. Итак, начнем со ссылок

• Репозиторий сервиса https://github.com/confluentinc/kafka-rest
• Репозиторий подсистемы 1С – https://github.com/vbondarevsky/Connector
• Докер образ REST сервиса для быстрой локальной отладки алгоритмов https://github.com/confluentinc/cp-all-in-one/tree/6.0.1-post/cp-all-in-one-community

Опять же напомню глоссарий – когда мы работаем с Apache Kafka мы должны помнить о семантике:

• Мы отправляем сообщения в потоковом режиме в виде записей разделов, которые сгруппированы в тему.
• После чего мы получаем записи из разделов темы по подписке – а в рамках подписки мы или сервер фиксируем/фиксирует смещение относительно начала журнала записей которую получил клиент-подписчик.

Указанный выше глоссарий нам понадобится, поэтому стоит перечитать его дважды – данная семантика нас будет преследовать почти на каждом шагу использования платформы Apache Kafka.

Перед стартом не забудьте выполнить команды подготовки кодирования

mkdir fund-kafka-experiments 

cd fund-kafka-experiments 

git clone https://github.com/confluentinc/cp-all-in-one

Для отладки нам понадобится запустить комплект платформы Kafka

cd cp-all-in-one/cp-all-in-one-communitydocker-compose up -d

Здесь я предполагаю, что у вас установлен Docker Engine – https://docs.docker.com/engine/install/

Обратите внимание – подключаться мы будем по порту 8082 на локальной машине… Для отладки работы нам этого будет достаточно.

Проверка без 1С: curl к REST Proxy

Прежде чем городить код на встроенном языке, полезно убедиться, что REST Proxy вообще жив и отвечает — это экономит часы отладки, когда потом что-то не работает в 1С и непонятно, где искать проблему: в 1С, в Docker или в самой Kafka. Отправка сообщения в тему jsontest:

curl -X POST -H "Content-Type: application/vnd.kafka.json.v2+json" \
  -H "Accept: application/vnd.kafka.v2+json" \
  --data '{"records":[{"value":{"foo":"bar"}}]}' \
  http://localhost:8082/topics/jsontest

В ответе будет назначенное смещение (offset) — именно его мы потом видим и в возвращаемом значении из 1С:

{"offsets":[{"partition":0,"offset":0,"error_code":null,"error":null}]}

Создание consumer-группы и подписка на тему:

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \
  --data '{"name": "my_consumer_instance", "format": "json", "auto.offset.reset": "earliest"}' \
  http://localhost:8082/consumers/my_json_consumer

curl -X POST -H "Content-Type: application/vnd.kafka.v2+json" \
  --data '{"topics":["jsontest"]}' \
  http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance/subscription

И чтение записей:

curl -X GET -H "Accept: application/vnd.kafka.json.v2+json" \
  http://localhost:8082/consumers/my_json_consumer/instances/my_consumer_instance/records

Если это отработало — дальше в 1С будет ровно тот же протокол, просто обёрнутый в HTTP-запросы через подсистему Connector или штатный объект HTTPЗапрос. И не забывайте закрывать consumer-инстанс, когда он больше не нужен (DELETE .../consumers/{group}/instances/{instance}) — иначе Kafka будет ждать heartbeat от «зависшего» клиента при ребалансировке группы, и остальные читатели в группе будут тормозить.

1с kafka. Самый простой отправитель

Итак, я предполагаю, что вы скачали подсистему для вызова HTTP сервисов (на сегодня актуальный релиз 2.1.3) и встроили в свою Конфигурацию получив общий модуль для вызова HTTP сервисов.

Как же в этом случае выглядит простой отправитель событий? Необходимо выполнить следующее:

Инициализировать массив записей

Инициализировать само событие

Добавить ЗАПИСЬ в массив ЗАПИСЕЙ

Преобразовать наши объекты в данные – в моем случае я использую JSON

Итак, данные готовы – остается только отправить, для этого мы указываем заголовок – чтобы REST Proxy Kafka понял какой формат данных мы стараемся передать.

И передаем:

Соответственно – мы вольны сами выбирать имя темы, объем передаваемых событий за один раз (размерность массива) и формат – JSON, XML, BASE64 и т.д.

Отдельно стоит отметить, что в возвращаемом значении хранится либо исключение/ошибка – либо смещение в потоке которое было назначено нашим записям, которые мы отправили. Собственно, это массив смещений.

И вот у вас теперь есть адаптер для 1С к серверу Кафка – без регистрации и смс. Фактически ваша задача превращается в творческий копипаст кода отсюда https://docs.confluent.io/platform/current/kafka-rest/quickstart.html

1с kafka. Группа получателей, получатель с автоматической фиксацией транзакцией

C получателем в экосистеме Kafka всегда сложней, здесь нам понадобится понимание того что Kafka это не RabbitMQ – имеется ввиду что у нас всегда есть не подписчик, а группа подписчиков. Предполагается, что «забиратели данных» будут делать это в многопоточном режиме. Поэтому поведение чуть сложней:

• Для начала нам понадобится функция, которая формирует правильные данные запроса

• Состав функции простой: мы явно указываем наш уникальный идентификатор клиента (потока/фонового задания), формат сообщения в котором мы его хотим принять и ключевое слово «самые ранние еще нами не полученные сообщения» 

• Теперь полученную структуру вместе с уже привычными заголовками мы отправляем в REST Proxy

Дальше есть особенность – уже связанная с отладочным запуском – дело в том что в объекте ВыданнаяПодписка нам выдан адрес внутри Docker DNS – и извне он по умолчанию не доступен, поэтому придется подменить в отладочный целях базовый URL

Итак – мы получили выданную подписку – то есть адрес ОТКУДА мы будем получать в нашем потоке сообщения из потока. Теперь надо указать на какой комплект ТЕМ мы подписываемся…

Внимательный читатель уже понял, что этот вызов также проходит через функции отправки управляющих параметров. А именно КаналПодписки() 

• Функция простая до невозможности – она формирует «список ТЕМ» на которые мы хотим подписаться.

Финальное мы должны начать получать записи  – здесь важно помнить про 3 ключевых момента:

• Стоит явно различать структуры «управление подпиской» и структуру управления «сериализацией» – те самые заголовки вызова
  • ЗаголовкиСериализацииЗаписей.Вставить(“Accept”, “application/vnd.kafka.json.v2+json”);
  • ЗаголовкиПодписки.Вставить(“Content-Type”, “application/vnd.kafka.json.v2+json”);
• По умолчанию вы получите ВСЕ записи с предыдущего подключения, поэтому вызывать получателя нужно как можно чаще. Это вам не librdkafka и NativeAPi компонента.
• Чтобы вы не делали – нужно явно удалить поток из группы подписчиков – чтобы не было проблем с перебалансировкой – поэтому и используется попытка.

И в итоге у вас теперь есть и подписчик и получатель. Как вы видите из примера кода – реальное сообщение хранится в значении VALUE – советую посмотреть на остальные поля: это метаданные, которые вам назначил потоковый сервер Kafka. Лучше всего это делать при помощи комбинации клавиш Shift+ F9 (имеется ввиду отладчик, точка остановки и просмотр  значения) 

Частые вопросы про Kafka REST Proxy и интеграцию с 1С

Почему v2 API, а не v3? В v2 сообщения кодируются через media-type вида application/vnd.kafka.json.v2+json (есть также avro.v2 и binary.v2), и только v2 поддерживает Kafka Headers — а это как раз то, что чаще всего нужно в интеграциях 1С: передать источник события, идентификатор корреляции, версию схемы и так далее. v3 API новее и его стоит использовать для административных операций — создания и удаления тем, изменения конфигурации кластера, но для продюсинга и потребления сообщений у v3 нет Headers, только обычный application/json. Для рабочей интеграции 1С — Kafka обычно достаточно v2, как и показано выше в примерах.

Обязателен ли Schema Registry? Нет. Если у вас уже есть контракты Avro или Protobuf между 1С и другими системами — REST Proxy умеет валидировать сообщения через Schema Registry. Для старта и для большинства интеграций 1С хватает обычного JSON без схемы, как в примерах этой статьи.

Что будет, если фоновое задание 1С зависнет и не удалит consumer-инстанс? Kafka ждёт heartbeat от участника группы согласно таймауту сессии, и пока этот таймаут не истёк, ребалансировка группы не произойдёт — остальные подписчики будут ждать освобождения партиций. Поэтому явное удаление инстанса из группы в примере получателя обёрнуто в попытку/исключение — это не паранойя, а обязательная часть протокола consumer-групп Kafka.

Kafka REST Proxy медленнее, чем нативный клиент (librdkafka)? Да, дополнительный HTTP-слой добавляет накладные расходы по сравнению с бинарным протоколом Kafka. Но для большинства интеграций 1С — где события формируются пользователем или регламентным заданием, а не миллионами в секунду — эта разница не критична, а простота эксплуатации (нет нативных компонент, ничего компилировать под платформу) обычно перевешивает.

В порядке завершения

Я сознательно сделал скриншоты кода – а не написал код для удобного копипаста. Я искренне надеюсь что вы сами как и я просто напишите этот код руками и проверите – насколько удобно и быстро можно вызывать любой REST сервис… Плюс – исходный код получают покупатели нашего вебинара по Apache Kafka и 1С, где мы намного больше рассмотрели кейсов и подходов к Apache Kafka

Ну а данная статья написана в качестве ответа тем, кто потратил время и деньги на написание своего велосипеда, что применительно к платформе Apache Kafka совершенно не нужно – так как основные затраты там на стороне поиска бизнес-кейсов, где бы она была применима. Писать что-то низкоуровневое для прикладного программиста – это совсем непрофильно. Тем более использование, как Вы видите, достаточно простое.

Похожие статьи

• 1С IIS сервер: установка IIS на Windows Server 2019
• Apache SSL: безопасная работа 1С в вебе
• Python 1C: микросервис на FastAPI