Брокеры сообщений и сквозная трассировка¶

Интеграции обмениваются данными через Kafka и брокеры очередей (AMQP, JMS) с помощью компонентов Camel. Ниже описано, как подключить брокер и как проследить один бизнес-объект через цепочку интеграций.

Kafka¶

Компонент camel-kafka есть в образе сборщика. Добавьте его в зависимости проекта:

camel.jbang.dependencies=...,camel-kafka

Адрес брокера и учётную запись храните в секрете Kubernetes:

apiVersion: v1
kind: Secret
metadata:
  name: integration-kafka
type: Opaque
stringData:
  KAFKA_BROKERS: "kafka.example.local:9093"
  KAFKA_SECURITY_PROTOCOL: "SASL_SSL"
  KAFKA_SASL_MECHANISM: "SCRAM-SHA-512"
  KAFKA_SASL_JAAS: >-
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="<пользователь>" password="<пароль>";

Подключите четыре ключа секрета переменными окружения в deployment.jkube.yaml, как в разделе Параметры и секреты, и сошлитесь на них в application.properties:

camel.component.kafka.brokers={{env:KAFKA_BROKERS}}
camel.component.kafka.securityProtocol={{env:KAFKA_SECURITY_PROTOCOL}}
camel.component.kafka.saslMechanism={{env:KAFKA_SASL_MECHANISM}}
camel.component.kafka.saslJaasConfig={{env:KAFKA_SASL_JAAS}}
events.topic=incoming.events
results.topic=processed.events
events.group=<имя интеграции>

SASL_SSL шифрует соединение. Если сертификат брокера выпущен корпоративным центром сертификации, интеграции нужен truststore с этим сертификатом (параметры camel.component.kafka.sslTruststoreLocation и camel.component.kafka.sslTruststorePassword). SASL_PLAINTEXT передаёт данные без шифрования, используйте его только там, где это разрешено.

Маршрут, который читает один топик и пишет в другой:

- route:
    id: events-consumer
    from:
      uri: kafka
      parameters:
        topic: "{{events.topic}}"
        groupId: "{{events.group}}"
        autoOffsetReset: earliest
      steps:
        - to:
            uri: kafka
            parameters:
              topic: "{{results.topic}}"

Что учесть при работе с Kafka:

Задавайте groupId по имени интеграции. Реплики одной интеграции с общим groupId делят разделы топика между собой. При сбое, перезапуске или перебалансировке группа может прочитать сообщение повторно.
Ключ записи определяет раздел топика. Консьюмер кладёт ключ входной записи в заголовок CamelKafkaKey, продюсер берёт ключ из того же заголовка. Порядок сообщений сохраняется только внутри раздела, поэтому в ключ удобно класть идентификатор бизнес-объекта: все его сообщения попадут в один раздел.
В Camel 4.21 строковые имена заголовков Kafka изменились: прежний kafka.KEY теперь называется CamelKafkaKey. После сборки проверьте ключ у записи в выходном топике, а не только заголовок в Camel.
Kafka может доставить сообщение повторно. Делайте обработку идемпотентной: например, отмечайте обработанные сообщения в базе и пропускайте повторы.
Если событие нужно отправить только после записи в базу, сохраните его в таблицу в той же транзакции, а в Kafka его опубликует отдельный маршрут (шаблон transactional outbox). Тогда при недоступном брокере событие не потеряется. Такой маршрут может отправить событие повторно, поэтому получатель всё равно должен отсеивать дубли.

Брокеры очередей: AMQP и JMS¶

Для AMQP 1.0 используется компонент camel-amqp, он тоже есть в образе сборщика. Фабрика соединений описывается бином проекта, адрес и учётная запись берутся из переменных окружения:

- beans:
    - name: jakarta.jms.ConnectionFactory
      type: org.apache.qpid.jms.JmsConnectionFactory
      properties:
        remoteURI: "{{env:AMQP_URL}}"
        username: "{{env:AMQP_USERNAME}}"
        password: "{{env:AMQP_PASSWORD}}"

Отправка в очередь:

- to:
    uri: amqp
    parameters:
      destinationType: queue
      destinationName: "{{notify.queue}}"

Чтение из очереди задаётся так же, в блоке from. Для адреса amqps:// интеграции нужно доверие к сертификату брокера. Брокеры с протоколом JMS подключаются компонентом camel-jms и фабрикой соединений из клиентской библиотеки брокера.

Сквозная трассировка¶

Каждая интеграция отправляет трейсы в Elastic APM (см. Трейсы). Внутри интеграции трасса непрерывная. Между интеграциями она продолжается, только если отправитель передал контекст трассировки в заголовках сообщения, а получатель его извлёк. Для HTTP и Kafka это зависит от компонентов и настроек, проверьте на своих маршрутах. Через промежуточную таблицу контекст не передаётся, и следующая интеграция начинает новую трассу.

Чтобы связать всю цепочку, используйте бизнес-ключ: идентификатор, который есть в каждом сообщении, например номер заказа, платежа или заявки.

В начале маршрута положите ключ в заголовок OrderId:
```
- setHeader:
    name: OrderId
    expression:
      simple:
        expression: ${body[id]}
```
Выражение работает, если тело уже преобразовано в Map, например шагом unmarshal из JSON. Из JSON-строки оно ключ не достанет.
Добавьте заголовок в MDC в application.properties:
```
camel.mdc.customHeaders=CAMEL_TRACE_ID,CAMEL_SPAN_ID,OrderId
```
Передавайте ключ дальше в теле или заголовках сообщения, чтобы следующая интеграция могла его взять.

log4j2.properties из комплекта поставки печатает ключ в строках обработки ([order=...]), а пайплайн Logstash записывает его в поле order.id. В Kibana вся цепочка находится одним запросом:

order.id: "<значение ключа>"

В этих строках есть и trace id, по нему открывается трасса в APM.

Имя заголовка можно изменить. Тогда поменяйте его в шаблоне appender.stdout.layout.pattern файла log4j2.properties и в фильтре пайплайна Logstash.