Kafka-ott-signature-interceptor#

Kafka-ott-signature-interceptor – модуль типа SIGNATURE (для контроля целостности сообщений), устанавливаемый на стороне потребителя. Находится в составе компонента EVTD в виде набора библиотек. Реализует интерфейсы ProducerInterceptor и ConsumerInterceptor, работает с типами String (например <String, String> (key, value)) или Array<Byte> (byte[]).

Предусловия#

В случае, если тип сообщения отличается от String или Array<Byte>, можно использовать kafka-certificate-signature-serde – аналогичная функциональность в виде де/сериализаторов.

ВАЖНО! На отправителе и получателе сообщения должен быть один и тот же интерсептор. Использовать, например, два интерсептора можно в случае, если у вас 2 продьюсера, 2 консьюмера и 2 топика – то есть все взаимодействия будут полностью разделены. У продьюсера/консьюмера может быть только один интерсептор, потому что если писать в один топик двумя разными продьюсерами с разными интерсепторами – при вычитке сообщения будут ошибки, так как механизм проверки целостности сообщения разный.

Последовательность выполнения#

Перехватчик собран и протестирован на версии kafka-клиента, которая указана в версии перехватчика в блоке <dependency>. Например:

<dependency>
  <groupId>ru.sbt.ss</groupId>
  <artifactId>kafka-ott-signature-interceptor_2.13</artifactId>
  <version>3.7.2-0.10.0</version>
</dependency>

, где 3.7.2 – версия Apache Kafka. Работа перехватчика с Spring Kafka возможна с версией 2.13.11 или ниже. Текущая версия поддерживает pem-форматы сертификатов.

ВАЖНО! Перехватчик Kafka-ott-signature-interceptor тянет за собой в качестве транзитивной зависимости библиотеку sbp.com.sbt.ott:ott-jsonrpc-hb-client:[version], которая не входит в состав компонента EVTD. Данная библиотека относится к серверной части ОТТ (ОТТ-сервис). Подробнее в документации продукта OTT (Platform V One-Time-Token) и его компонента OTTS.

Соотношение между Kafka-ott-signature-interceptor из компонента EVTD и компонентом OTTS продукта OTT (Platform V One-Time-Token)#

При подключении kafka-ott-signature-interceptor для ОТТ-подписи потока событий важно понимать, что используется ДВА продукта:

  • продукт Platform V One-Time-Token (OTT), компонент OTTS – предоставляет функциональность OTT-подписи;

  • компонент EVTD – предоставляет библиотеки класса-перехватчика, который позволяет подключить OTT-клиент к kafka-клиенту и сконфигурировать его.

Т.е. внутри себя Kafka-ott-signature-interceptor транзитивно подтягивает ОТТ-клиент, конфигурирует его теми настройками, которые ему передали, и управляет его запуском/остановкой.

Логика работы:

  1. Producer-interceptor запрашивает токен у централизованного сервиса ОТТ.

  2. Producer-interceptor подписывает тело сообщения (record.value()) с помощью токена.

  3. Producer-interceptor добавляет подпись (в Base64) и токен (в формате jws) в заголовки сообщения.

  4. Consumer-interceptor получает токен (jws) из заголовка сообщения и проверяет его валидность.

  5. Consumer-interceptor получает подпись из заголовка сообщения и проверяет его валидность с помощью токена.

  6. В случае, если сообщение не прошло проверку подписи при получении (consumer.poll()), поведение настраивается параметром interceptor.signature.mode:

    • failOnValue (используется по умолчанию) — клиент получит сообщение-заглушку вместо невалидного сообщения, которое содержит null вместо value и выбросит исключение при вызове record.value();

    • filter — сообщение об ошибке будет залогировано в error, клиент не получит невалидное сообщение;

    • failOnConsume — метод consumer.poll() выбросит исключение ru.sbt.ss.kafka.interceptors.ConsumerInterceptorError, клиент не получит ни одного сообщения из пачки.

Используемые методы:

Название интерфейса

Метод

Описание

public interface ProducerInterceptor

configure

Вызывается при создании Kafka-клиента, создает и конфигурирует OTT-клиент

public interface ProducerInterceptor

onSend

Вызывается перед отправкой каждого сообщения (подписывает сообщение, используя технологию OTT

public interface ProducerInterceptor

close

Закрывает OTT-клиент

public interface ConsumerInterceptor

configure

Вызывается при создании Kafka-клиента, создает и конфигурирует OTT-клиент

public interface ConsumerInterceptor

onConsume

Вызывается после получения сообщений, непосредственно перед передачей сообщений в прикладной код (проверяет OTT-токен, проверяет подпись сообщения с помощью технологии OTT)

public interface ConsumerInterceptor

close

Закрывает OTT-клиент

ВАЖНО! При использовании Kafka-ott-signature-interceptor возможно падение производительности передачи событий (более, чем в 10 раз в отдельных случаях). Для установления конкретных показателей для определенного взаимодействия необходимо проведение НТ.

Подключение к Kafka-клиентам#

  1. Добавить актуальную версию интерсептора в зависимости проекта.

  2. Сконфигурировать kafka-клиенты в соответствии с примерами.

Использование в cloud-среде#

  1. Добавить зависимость интерсептора в сборку образа приложения.

  2. Сконфигурировать kafka-клиенты в соответствии с разделом Развертывание в облачной среде.

Подключение к Spring kafka#

Для настройки kafka-клиента в Spring необходимо выполнить следующие шаги:

1. Настройка KafkaTemplate и ProducerFactory

  @Configuration
  public static class JsonConfigConsumer {

    @Bean
    public ProducerFactory<String, String> producerFactorySettingsFromFile() {

      /**
        Указываем путь до параметров `PATH_TO_PRODUCER_VALID_PROPERTIES`, дополнительные параметры можно передать через `Map.of("nameParam, valueParam")`
      */
      final var producerProperties = loadProperties(PATH_TO_PRODUCER_VALID_PROPERTIES, Map.of());

      return new DefaultKafkaProducerFactory<>(propertiesToMapConverter(producerProperties));
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplateSettingsFromFile(ProducerFactory<String, String> producerFactorySettingsFromFile) {
      return new KafkaTemplate<>(producerFactorySettingsFromFile);
    }

    private Properties loadProperties(String path, Map<String, String> additionalSettings) {
      try {
        return FileUtils.loadPropertiesFromClasspath(path, additionalSettings);
      } catch (IOException e) {
      throw new RuntimeException(e);
      }
    }
  }

2. Настройка Listeners и ConsumerFactory

  @Configuration
  public static class JsonConfigConsumer {

        @Bean
        public ConsumerFactory<String, String> consumerFactorySettingsFromFile() {
            /**
              Указываем путь до параметров `PATH_TO_CONSUMER_VALID_PROPERTIES`, дополнительные параметры можно передать через `Map.of("nameParam, valueParam")`
            */
            Properties consumerProperties = loadProperties(PATH_TO_CONSUMER_VALID_PROPERTIES, Map.of());
            return new DefaultKafkaConsumerFactory<>(propertiesToMapConverter(consumerProperties));
        }

        @Bean
        public KafkaListenerContainerFactory<
                ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactorySettingsFromFile(
                ConsumerFactory<String, String> consumerFactorySettingsFromFile,
                ErrorHandler createErrorHandler
        ) {
            ConcurrentKafkaListenerContainerFactory<String, String> factory =
                    new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactorySettingsFromFile);

            return factory;
        }
  }

3. Настройка Listener

  @EnableKafka
  @Component
  public class ListenersKafkaConsumerJson {

    @KafkaListener(
            id = "id-1",
            topics = "NAME_TOPIC",
            groupId = "group-id",

            /**
              Необходимо указать имя bean фабрики с нужными параметрами.
            */
            containerFactory = "kafkaListenerContainerFactorySettingsFromFile",
    )
    private void listenerConfigurationFile(String data) {
      try {
        JsonConsumerSpringTest.RESPONSE_QUEUE.put(data);
      } catch (InterruptedException e) {
            throw new RuntimeException(e);
      }
    }
  }

4. Настройка ErrorHandler (опционально)

Чтобы задать логику обработки ошибок, необходимо при создании KafkaListenerContainerFactory задать собственную реализацию ErorHandler, затем передать в метод setErrorHandler(errorHandler).

Пример реализации ErrorHandler:

new ErrorHandler() {
  @Override
  public void handle(Exception e, ConsumerRecord<?, ?> consumerRecord) {
     throw e;
  }
};

Конфигурация ОТТ-запроса#

Для OttSignatureProducerInterceptor атрибуты ott.request.* отправляются на сервер ОТТ для авторизации запроса на создание токена. Некоторые параметры попадают в токен в виде token claims.

Для OttSignatureConsumerInterceptor атрибуты ott.request.* используются для проверки токена в заголовке сообщения. Все атрибуты, указанные в конфигурации считаются обязательными, если они отсутствуют в токене — токен не пройдет валидацию и подпись будет считать невалидной.

Настройки OTT-запроса ott.request.* транслируются в вызовы методов builder com.sbt.ott.api.client.TokenRequestBuilder OTT-клиента.

Атрибуты ott.request.add.* в конфигурации консьюмера должны быть указаны как у продюсера для корректной валидации параметров в подписи события.

Строковые параметры#

Параметр конфигурации

Метод builder

ott.request.invoker = testInvoker

builder.invoker("testInvoker")

ott.request.receiver = testReceiver

builder.invoker("testReceiver")

ott.request.url = /test/url

builder.url("/test/url")

ott.request.domain = testDomain

builder.domain("testDomain")

ott.request.action = testAction

builder.action("testAction")

Фиксированные составные параметры (для раздела Создание атрибутов)#

Могут состоять из опциональной категории, имени параметра и его значения, разделитель по умолчанию #.

!ВАЖНО Атрибуты ott.request.add.* в конфигурации консьюмера должны быть указаны как у продюсера для корректной валидации параметров в подписи события.

Параметр конфигурации

Метод

ott.request.add.subject = testSubjectAttributeName#testSubjectAttributeValue

builder.addSubject(BasicAttribute("testSubjectAttributeName", "testSubjectAttributeValue"))

ott.request.add.recipient.subject = testRecipientAttributeName#testRecipientAttributeValue

builder.addRecipientSubject(BasicAttribute("testRecipientAttributeName", "testRecipientAttributeValue"))

ott.request.add.action = testActionAttributeName#testActrionAttributeValue

builder.addAction(BasicAttribute("testActionAttributeName", "testActionAttributeValue"))

ott.request.add.resource = testResourceAttributeName#testResourceAttributeValue

builder.addResource(BasicAttribute("testResourceAttributeName", "testResourceAttributeValue"))

ВАЖНО! В параметр ott.request.add.resourceподставляется url, куда идет запрос на авторизацию. Например:

interceptor.signature.ott.request.add.resource = urn:sbrf:names:pprb:1.0:api:interface:fullname

Еще пример из логов:

00:22:22.878 [INFO ] [c.s.o.b.a.s.impl.TokenServiceImpl] [ForkJoinPool-1-worker-19] - Запрос на генерацию токена для вызова [JwsTokenCreateReq{tokenClaims=JwsTokenClaims(invoker=testModuleA, receiver=null, url=/api/v3/GetSalaryAgreementInfo, urlEvent=null, domain=null, encKeyParam=null, action=request, anonymous=null, tokenType=null, reqId=68601979-82df-4cd0-a019-7982dfecd057), namedAttrs={Attributes=[id=urn:sbrf:names:pprb:1.0:api:interface:fullname; category=urn:oasis:names:tc:xacml:3.0:attribute-category:resource; type=http://www.w3.org/2001/XMLSchema#string; values=[/api/v3/GetSalaryAgreementInfo], id=urn:sbrf:names:pprb:1.0:module:id; category=urn:oasis:names:tc:xacml:1.0:subject-category:access-subject; type=http://www.w3.org/2001/XMLSchema#string; values=[ci03155960-rspp-pprb-prom], id=urn:sbrf:names:pprb:1.0:action:id; category=urn:oasis:names:tc:xacml:3.0:attribute-category:action; type=http://www.w3.org/2001/XMLSchema#string; values=[request], id=urn:sbrf:names:pprb:1.0:req:type; category=urn:oasis:names:tc:xacml:3.0:attribute-category:environment; type=http://www.w3.org/2001/XMLSchema#string; values=[consumer], id=ott:realm; category=urn:oasis:names:tc:xacml:3.0:attribute-category:environment; type=http://www.w3.org/2001/XMLSchema#string; values=[mmt]]}}] отправлен на сервер ОТТ

Здесь значение /api/v3/GetSalaryAgreementInfo, оно же в url=/api/v3/GetSalaryAgreementInfo

Прочие составные параметры (для раздела Создание атрибутов)#

Задаются аналогично фиксированным составным параметрам, можно задать несколько параметров из каждой категории ott.request.add.environment или ott.request.add.attribute.

!ВАЖНО Атрибуты ott.request.add.* в конфигурации консьюмера должны быть указаны как у продьюсера для корректной валидации параметров в подписи события.

Параметр

Метод

ott.request.add.environment.* = testEnvAttributeName#testEnvAttributeValue

builder.addEnvironment(BasicAttribute("testEnvAttributeName", "testEnvAttributeValue"))

ott.request.add.attribute.* = testAttributeName#testAttributeValue

builder.addAttribute(BasicAttribute("testAttributeName", "testAttributeValue"))

Параметры ott.request.* для продьюсера отвечают за то, с какими параметрами запросить токен:

producer.interceptor.signature.ott.request.add.action=urn:sbrf:names:pprb:1.0:action:id#request - добавить параметр 'action.id = request' в запрос на выпуск токена

Параметры ott.request.* для consumer отвечают за проверку токена:

consumer.interceptor.signature.ott.request.add.action=urn:sbrf:names:pprb:1.0:action:id#request - проверить параметр 'action.id = request' в токене

Имена параметров после ott.request.add.environment. и ott.request.add.attribute. игнорируются, но должны быть уникальны для корректной передачи всех параметров в конфигурацию.

Пример конфигурации прочих параметров:

ott.request.add.environment.env1 = testEnvAttributeName1#testEnvAttributeValue1
ott.request.add.environment.env2 = testEnvAttributeName2#testEnvAttributeValue2
ott.request.add.attribute.1 = testAttributeName1#testAttributeValue1
ott.request.add.attribute.2 = testAttributeName2#testAttributeValue2

Создание атрибутов#

// Для составного параметра со значением ott.request.add.subject = subjectCategory#subjectId#subjectValue
var attribute = new BasicAttribute();

attribute.setCategory("subjectCategory");
attribute.setId("subjectId");
attribute.addValue("subjectValue");
attribute.setMandatory(true);

Примеры конфигурации#

Настройки Kafka Producer и Kafka Consumer конфигурируются там же, где и настройки Kafka-клиента.

Пример конфигурации Kafka Producer#

# 1) Подключить interceptor
interceptor.classes = ru.sbt.ss.kafka.ott.OttSignatureProducerInterceptor

# 2) Настроить ОТТ клиент

# Id модуля (также alias клиентского сертификата)
interceptor.signature.ott.module.id = ott-test-moduleA

# Хосты ОТТ
interceptor.signature.ott.service.url = https://ott-service.otts-std-ift2.ingress.apps.ocp.ift-02.solution.sbt
interceptor.signature.ott.service.hosts = { IP_ADDRESS }

# Realm авторизации
interceptor.signature.ott.authz.realm = synapse

# ОПЦИОНАЛЬНО закрывать ott-клиент при закрытии kafka-клиента
# Библиотека ott кеширует клиенты по module.id, но не удаляет их из кэша при закрытии,
# из-за чего новый (или перезапустившийся) kafka-клиент может получить из кеша уже закрытый ott-клиент
# По умолчанию ott-клиент не закрывается
# interceptor.signature.ott.close.client = false

# 3) Настроить параметры запроса к ОТТ-сервису
# Атрибуты `*.ott.request.*` отправляются на сервер отт для авторизации запроса на создание токена. Некоторые параметры попадают в токен в виде token claims.

# ОПЦИОНАЛЬНО настроить разделитель строки составного параметра
# interceptor.signature.ott.attribute.delim = #

# 3.1) Пример настройки для realm mmt (целевой на момент написания документации):
interceptor.signature.ott.request.add.subject = urn:sbrf:names:pprb:1.0:module:id#ott-test-moduleA
interceptor.signature.ott.request.add.resource = urn:sbrf:names:pprb:1.0:api:interface:fullname
interceptor.signature.ott.request.add.environment.realm = ott:realm#mmt
interceptor.signature.ott.request.add.action = urn:sbrf:names:pprb:1.0:action:id#request

# 3.2) Пример настройки для realm synapse (legacy):
# В версиях < 0.3.0 присутствовали специфичные параметры для realm synapse, из которых строился атрибут ott:synapse:uri
# 1: interceptor.signature.ott.synapse.uri.prefix = /cluster/
# 2: interceptor.signature.ott.synapse.uri.publish.suffix = /publish


# interceptor.signature.ott.request.add.subject = ott:app#moduleId
# interceptor.signature.ott.request.add.environment.realm = ott:realm#synapse

# 4) Настроить TLS для ОТТ-клиента
# 4.1) Пример настройки с сертификатами в jks-хранилище

# alias серверного сертификата отт в truststore
interceptor.signature.ott.service.cert.alias = ott-service

# Тип хранилища сертифкатов (по умолчанию pkcs12)
interceptor.signature.ott.certstore.type = JKS

# Путь до хранилища клиентских сертификатов
interceptor.signature.ott.certstore.path = ssl/keystore.jks

# Путь до хранилища доверенных сертификатов
interceptor.signature.ott.trust.store.path = ssl/truststore.jks

# Пароль от хранилища клиентских сертификатов
interceptor.signature.ott.certstore.pwd = password

# Пароль от приватного ключа
interceptor.signature.ott.certstore.private.key.pwd = password

# Пароль от хранилища доверенных сертификатов
interceptor.signature.ott.trust.store.pwd = password

# Версия TLS протокола
interceptor.signature.ott.client.tls = TLSv1.2

# 4.2) Пример настройки с сертификатами в формате pem

# alias серверного сертификата отт в truststore
interceptor.signature.ott.service.cert.alias = ott-service

# Тип хранилища сертифкатов (по умолчанию pkcs12)
interceptor.signature.ott.type = PEM
interceptor.signature.ott.certstore.type = PEM

# Сертификат сервиса ОТТ
interceptor.signature.ott.service.crt  = ott-service.pem

# Сертификат УЦ, подписавшего сертификат сервиса ОТТ
interceptor.signature.ott.service.tls.crt = ott-service-tls.pem

# Сертификат клиента
interceptor.signature.ott.client.crt = ott-client.pem

# Приватный ключ клиента
interceptor.signature.ott.client.private.key = ott-ckient-private-key.key

# Версия TLS протокола
interceptor.signature.ott.client.tls = TLSv1.2

# 5) ОПЦИОНАЛЬНО Настроить имена системных заголовков
# Имя заголовка с подписью сообщения
# interceptor.signature.header = signature

# Префикс для имен системных заголовков
# interceptor.signature.attributes.prefix = signature.

# Имя заголовка с токеном
# interceptor.signature.ott.token.header = token

# 6) ОПЦИОНАЛЬНО
# Включить логирование интерсептора при подписи сообщений
# interceptor.trace.logger.enabled = true

# Задать имя логгера
# По умолчанию генерируется автоматически в виде <Имя_класса_интерсептора>[<client.id_из_настроек_kafka_клиента>], например:
# ru.sbt.ss.kafka.interceptors.CertificateSignatureConsumerInterceptor[Consumer-1]
# Если client.id не задан - он генерируется автоматически
# interceptor.trace.logger.name = LoggerName

# 7) ОПЦИОНАЛЬНО
# Включить публикацию jmx-метрик количества успешно и неудачно обработанных сообщений
# interceptor.jmx.metrics.enabled = true

# 8) ОПЦИОНАЛЬНО

# Включить публикацию телеметрии
# interceptor.telemetry.enabled = true

# Задать путь к файлу конфигурации телеметрии
# interceptor.telemetry.config.path = /path/to/config/file

# Включить отображение спанов подписи/проверки подписи сообщений (по-умолчанию false)
# interceptor.telemetry.callback.enabled = true

Для использования телеметрии подключите библиотеку kafka-telemetry-interceptor

Пример конфигурации Kafka Consumer#

# 1) Подключить перехватчик
interceptor.classes = ru.sbt.ss.kafka.ott.OttSignatureConsumerInterceptor

# 2) Настроить ОТТ клиент

# Id модуля (также alias клиентского сертификата)
interceptor.signature.ott.module.id = ott-test-moduleA

# Хосты ОТТ
interceptor.signature.ott.service.url = https://ott-service.otts-std-ift2.ingress.apps.ocp.ift-02.solution.sbt
interceptor.signature.ott.service.hosts = { IP_ADDRESS }

# Realm авторизации
interceptor.signature.ott.authz.realm = synapse

# ОПЦИОНАЛЬНО закрывать ott-клиент при закрытии kafka-клиента
# Библиотека ott кеширует клиенты по module.id, но не удаляет их из кэша при закрытии,
# из-за чего новый (или перезапустившийся) kafka-клиент может получить из кеша уже закрытый ott-клиент
# По умолчанию ott-клиент не закрывается
# interceptor.signature.ott.close.client = false

# 3) Настроить параметры запроса к ОТТ-сервису
# Атрибуты `ott.request.*` используются для проверки токена в заголовке сообщения
# Атрибуты `ott.request.add.*` должны быть указаны как у producer для корректной валидации параметров в подписи события
# Все атрибуты, указанные в конфигурации считаются обязательными, если они отсутствуют в токене - токен не пройдет валидацию и подпись будет считать невалидной.

# ОПЦИОНАЛЬНО настроить разделитель строки составного параметра
# interceptor.signature.ott.attribute.delim = #

# 3.1) Пример настройки для realm mmt (целевой на момент написания документации):
interceptor.signature.ott.request.add.subject = urn:sbrf:names:pprb:1.0:module:id#ott-test-moduleA
interceptor.signature.ott.request.add.resource = urn:sbrf:names:pprb:1.0:api:interface:fullname
interceptor.signature.ott.request.add.environment.realm = ott:realm#mmt

# 3.2) Пример настройки для realm synapse (legacy):
# В версиях < 0.3.0 присутствовали специфичные параметры для realm synapse, из которых строился атрибут ott:synapse:uri:
# 1: interceptor.signature.ott.synapse.uri.prefix = /cluster/
# 2: interceptor.signature.ott.synapse.uri.publish.suffix = /publish


# interceptor.signature.ott.request.add.subject = ott:app#moduleId
# interceptor.signature.ott.request.add.environment.realm = ott:realm#synapse

# 4) Настроить TLS для ОТТ-клиента

# 4.1) Пример настройки с сертификатами в jks-хранилище

# alias серверного сертификата отт в truststore
interceptor.signature.ott.service.cert.alias = ott-service

# Тип хранилища сертифкатов (по умолчанию pkcs12)
interceptor.signature.ott.certstore.type = JKS

# Путь до хранилища клиентских сертификатов
interceptor.signature.ott.certstore.path = ssl/keystore.jks

# Путь до хранилища доверенных сертификатов
interceptor.signature.ott.trust.store.path = ssl/truststore.jks

# Пароль от хранилища клиентских сертификатов
interceptor.signature.ott.certstore.pwd = password

# Пароль от приватного ключа
interceptor.signature.ott.certstore.private.key.pwd = password

# Пароль от хранилища доверенных сертификатов
interceptor.signature.ott.trust.store.pwd = password

# Версия TLS протокола
interceptor.signature.ott.client.tls = TLSv1.2

# 4.2) Пример настройки с сертификатами в формате pem

# alias серверного сертификата отт в truststore
interceptor.signature.ott.service.cert.alias = ott-service

# Тип хранилища сертифкатов (по умолчанию pkcs12)
interceptor.signature.ott.type = PEM
interceptor.signature.ott.certstore.type = PEM

# Сертификат сервиса ОТТ
interceptor.signature.ott.service.crt  = ott-service.pem

# Сертификат УЦ, подписавшего сертификат сервиса ОТТ
interceptor.signature.ott.service.tls.crt = ott-service-tls.pem

# Сертификат клиента
interceptor.signature.ott.client.crt = ott-client.pem

# Приватный ключ клиента
interceptor.signature.ott.client.private.key = ott-ckient-private-key.key

# Версия TLS протокола
interceptor.signature.ott.client.tls = TLSv1.2

# 5) ОПЦИОНАЛЬНО Настроить удаление системных заголовков из сообщения
# interceptor.signature.remove.headers = false

# 6) ОПЦИОНАЛЬНО Настроить режим работы консьюмера при ошибках (по умолчанию failOnValue)
# interceptor.signature.mode = failOnValue

# 7) ОПЦИОНАЛЬНО Настроить имена системных заголовков
# Имя заголовка с подписью сообщения
# interceptor.signature.header = signature

# Префикс для имен системных заголовков
# interceptor.signature.attributes.prefix = signature.

# Имя заголовка с токеном
# interceptor.signature.ott.token.header = token

# 8) ОПЦИОНАЛЬНО
# Включить логирование интерсептора при подписи сообщений
# interceptor.trace.logger.enabled = true

# Задать имя логгера
# По умолчанию генерируется автоматически в виде <Имя_класса_интерсептора>[<client.id_из_настроек_kafka_клиента>], например:
# ru.sbt.ss.kafka.interceptors.CertificateSignatureConsumerInterceptor[Consumer-1]
# Если client.id не задан - он генерируется автоматически
# interceptor.trace.logger.name = LoggerName

# Включить публикацию jmx-метрик количества успешно и неудачно обработанных сообщений
# interceptor.jmx.metrics.enabled = true

# 9) ОПЦИОНАЛЬНО

# Включить публикацию телеметрии
# interceptor.telemetry.enabled = true

# Задать путь к файлу конфигурации телеметрии
# interceptor.telemetry.config.path = /path/to/config/file

# Включить отображение спанов подписи/проверки подписи сообщений (по-умолчанию false)
# interceptor.telemetry.callback.enabled = true

Для использования телеметрии подключите библиотеку kafka-telemetry-interceptor

Относительные пути в конфигурации#

Относительные пути до файлов в конфигурации обычно разрешаются относительно директории запуска приложения.

В некоторых случаях (для автоматических установок, например, для установок с помощью скриптов) может быть полезно явно указать root-директорию, относительно которой будут разрешаться относительные пути:

  • interceptor.config.root.dir (общая для всех перехватчиков с подобной настройкой);

  • interceptor.signature.ott.config.root.dir (переопределяет общую настройку).

Работает только для настроек interceptor.signature.ott.certstore.path и interceptor.signature.ott.trust.store.path.

Пример конфигурации:

interceptor.signature.ott.config.root.dir=/full/path/to
# interceptor.config.root.dir=/full/path/to

# /full/path/to/ssl/keystore.jks
interceptor.signature.ott.certstore.path = ssl/keystore.jks

# /full/path/to/ssl/truststore.jks
interceptor.signature.ott.trust.store.path = ssl/truststore.jks

JMX метрики#

При включении публикации JMX-метрик с помощью настройки interceptor.jmx.metrics.enabled = true в JMX будут добавлены метрики с количеством успешно и ошибочно обработанных сообщений и информацией о подключенном interceptor:

kafka.producer:type=producer-interceptor-metrics,client-id=<client-id>,interceptor=CertificateSignatureInterceptor,name=FailedProcessedMessage
kafka.producer:type=producer-interceptor-metrics,client-id=<client-id>,interceptor=CertificateSignatureInterceptor,name=SuccessfulProcessedMessage
kafka.producer:type=producer-interceptor-metrics,client-id=<client-id>,interceptor=CertificateSignatureInterceptor,name=Info

или

kafka.consumer:type=consumer-interceptor-metrics,client-id=<client-id>,interceptor=CertificateSignatureInterceptor,name=FailedProcessedMessage
kafka.consumer:type=consumer-interceptor-metrics,client-id=<client-id>,interceptor=CertificateSignatureInterceptor,name=SuccessfulProcessedMessage
kafka.consumer:type=consumer-interceptor-metrics,client-id=<client-id>,interceptor=CertificateSignatureInterceptor,name=Info

, где client-id берется из конфигурации client.id, при отсутствии в конфигурации генерируется автоматически.

Результат#

Выполнено подключение перехватчика Kafka-ott-signature-interceptor.