На NiFi как отличить сообщения, приходящие от двух разных какфа?

Question

У меня есть два разных приложения с именами А1 и А2.Каждое приложение имеет свой собственный сервер Kakfa.Сообщения, поступающие с этих двух серверов (брокеров) KAKFA, отправляются на NiFi.

Каждый Kakfa имеет разные названия тем, и на основании этого я могу различать сообщения, приходящие от Kakfa.Но кроме названия темы Кафки, есть ли в NiFi другие способы различать сообщения, приходящие от двух разных Kakfa?Есть ли у них какой-нибудь процессор NiFi, который проверяет название темы, а затем решает выбрать следующий маршрут?

Большое спасибо.С уважением, Йешван

Answer 1

Другое разрешение.Для этой цели вы можете использовать процессор CryptographicHashContent.Вы получаете уникальный хэш-код содержимого и на основании этого можете определить свое сообщение.

Примечание: Процессор HashContent помечен как устаревший с Apache NiFi1.8.0 из-за нестандартной устаревшей реализации и может быть удален в следующих версиях.

Answer 2

Если вы используете процессоры NiFi Kafka (ConsumeKafka / ConsumeKafkaRecord) для получения сообщений от Kafka, они будут выводить сообщения в виде FlowFiles.Они содержат один атрибут с именем kafka.topic, который будет содержать название темы, из которой пришло сообщение.

Для маршрутизации сообщений на основе имени темы вы можете использовать процессор RouteOnAttribute.Например, у вас есть две темы topicA & topicB.Затем необходимо настроить процессор RouteOnAttribute следующим образом:

Затем подключитеОтношение topic-a & topic-b к отдельным потокам на основе ваших требований.Если вы добавляете больше источников Kafka, все, что вам нужно сделать, это обновить RouteOnAttribute еще одним динамическим отношением.Пример: topic-c : ${kafka.topic:equals('topicC')}