Spring Cloud Kafka Streams Dynami c Преобразование сообщений на основе информации заголовка

Question

Я пытаюсь использовать Spring Cloud Kafka Streams для обработки сообщений от Kafka Topi c, который содержит сообщения различных типов. Например, мы получаем сообщение JSON от topi c, которое может быть сообщением типа A или типа B. Производитель добавляет тип сообщения в заголовок, есть ли способ прочитать эту информацию заголовка в Functional Binder и соответствующим образом преобразовать сообщение? Или также есть опция «Выбор» для ветвления при поступлении сообщений, для направления сообщения нужному конвертеру?

Answer 1

Если вы сконфигурируете привязку для использования nativeDecoding, десериализация выполняется Kafka (через свойство потребителя value.deserializer).

spring-kafka предоставляет JsonDeserializer, который ищет информацию о типе в Speci c заголовки (устанавливаются соответствующими JsonSerializer.

. Он также предоставляет DelegatingDeserializer, который позволяет вам выбрать, какой десериализатор использовать на основе значения в заголовке spring.kafka.serialization.selector.

См. Пружина для Apache Справочное руководство Kafka для получения дополнительной информации.