добавить настраиваемый показатель задержки на уровне записи c в потоках kafka

Question

Я пытаюсь добавить специфику c metri c в свое приложение kafka-streams, которое будет измерять задержку и сообщать в jmx. Я использую StreamsDSL в scala, поэтому использование ProcessorAPI для показателей (которые, как я знаю, возможны) для меня не сработает.

основные c вещи, которые я хотел бы понять, это :

1. как извлечь указанные c свойства записи (т.е. заголовки) для использования как часть метрич. c расчет
2. Как добавить новые метри c в метрики сообщаются jmx

Спасибо!

Answer 1

Вам нужно будет вернуться к Processor API для доступа к метаданным записи, таким как заголовки, и для регистрации пользовательских метрик.

Обратите внимание на мысль, что вы можете комбинировать DSL и Processor API, поэтому переходить с DSL не обязательно. Вместо этого вы можете подключить пользовательский Processors или Transformers через KStream.process() или KStream.transform() (обратите внимание, что есть несколько «братьев и сестер» к transform(), которые вы можете использовать вместо transform()).