Kafka

Общие сведения

Название плагина: kafka
Ссылка для загрузки плагина: скачать Обновить ссылку
Минимальная версия Monq: 9.0.0
Минимальная версия агента Monq: 3.0.0

Описание:
Плагин реализует непрерывное потоковое соединение с брокером сообщений Apache Kafka для получения данных из различных топиков.

Может быть использован только при настройке Рабочей конфигурации агента

Формат сценария рабочей конфигурации агента

continuous-jobs:
  - init-step:
      plugin: kafka
      with:
        brokers:
          - 192.168.1.3:9092
          - 192.168.1.4:9092
        brokersConnectionTimeout: 5
        brokersConnectionRetriesInterval: 10
        heartbeatsInterval: 3
        sessionTimeout: 45
        topics:
          - example_topic
        clientId: monq-agent
        groupId: monq-agent
        format: json
        pullMode: current
        pollIntervalMs: 500
        chunkSize: 32768
        bufferType: memory
        streamKey: <your_stream_key>
        bufferInputName: buffer-1
        labels:
          label1: value1
          label2: value2 
      with-secured:
        protocol: plaintext
        sasl.mechanism: PLAIN
        sasl.username: <username>
        sasl.password: <password>

Параметры конфигурации используемые в сценарии

Параметр	Тип	Обязательный параметр	Значение по умолчанию	Возможные значения	Описание
brokers	string (array)	да	-		Список брокеров Kafka. Обязательный параметр
brokersConnectionTimeout	integer	нет	5 sec		Таймаут подключения к брокерам (в секундах)
brokersConnectionRetriesInterval	integer	нет	10		Пауза между провалившимися попытками подключения (в секундах). При неудачном подключении пауза между попытками удваивается после каждой следующей попытки (экспоненциальная задержка).К каждой паузе также будет прибавляться время таймаута подключения (brokersConnectionTimeout).Каждая неудачная попытка логируется с уровнем Error
heartbeatsInterval	integer	нет	3 sec		Период отправки heartbeats (в секундах)
sessionTimeout	integer	нет	45 sec		Настройка таймаута сессии - время, за которое сессия не будет оборвана в случае отсутствия heartbeat (в секундах)
topics	string (array)	да	-		Список топиков, которые нужно прочесть. Числовое значение (<=16)
clientId	string	нет	-		Идентификатор агента в системе Kafka
groupId	string	нет	Название агента (статического)		Параметр позволяющий задать название группе потребителей в системе Kafka для распределенного чтения сообщений в случае подключения дополнительных агентов в эту же группу
format	string	нет	json	json, raw	Ожидаемый формат получения данных
pullMode	string	нет	current	all, current, forceAll	Режим получения сообщений. Значения: current — только новые сообщения, история пропускается (офсет сохраняется); all — дочитывает всю историю с последнего сохранённого офсета, затем ждёт новые (офсет сохраняется); forceAll — при каждом запуске читает всё с начала топика (офсет не сохраняется).
pollIntervalMs	integer	нет	500 ms		Интервал опроса брокеров Kafka (в миллисекундах)
chunkSize	integer	нет	32768	Любое число	Задает максимальный размер "порций" данных для хранения входящих сообщений (в байтах)
bufferType	string	нет	filesystem	memory, filesystem	Определяет тип памяти для очереди. Доступно: memory - память агента, filesystem - на диске.
streamKey	string	да	-		Ключ потока, в который необходимо отправлять собранные данные
bufferInputName	string	нет	Динамическое название		Параметр необходим если пользователь хочет с одного агента подключаться к разным брокерам Kafka
labels	object	-			Дополнительные поля, которые будут добавлены к результирующим событиям
extraConfig	object	нет	-		Расширенные параметры конфигурации, доступные в используемой библиотеке Confluent.Kafka.
with-secured	object	нет	-		Настройки безопасности (функционал агента)
with-secured.protocol	string	нет	plaintext	plaintext, sasl_plaintext, ssl, sasl_ssl	Протокол безопасности, используемый для связи с брокерами. Значения: plaintext - без аутентификации и шифрования; sasl_plaintext - с аутентификацией, без шифрования; ssl - без аутентификации, но с шифрованием; sasl_ss - с аутентификацией и шифрованием.
with-secured.sasl.mechanism	string	нет	PLAIN	PLAIN, SCRAM-SHA-256, SCRAM-SHA-512	Механизм SASL для аутентификации
with-secured.sasl.username	string	да (при использовании протокола sasl_plaintext)	-		Имя пользователя SASL для использования с механизмами PLAIN и SASL-SCRAM
with-secured.sasl.password	string	да (при использовании протокола sasl_plaintext)	-		Пароль SASL для использования с механизмами PLAIN и SASL-SCRAM

Параметр brokers

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

Пример:

{
  brokers:
          - 192.168.1.3:9092
          - 192.168.1.4:9092
}

Если в параметре задано больше одного брокера плагин будет пробовать подключиться к первому в списке. В случае неудачи - ко второму, и так далее.

При неудачном подключении плагин генерирует ошибку.

Таймаут подключения к брокерам настраивается пользователем.

Пример:

{
  brokersConnectionTimeout: 5
}