Question

У меня проблемы с попыткой десериализации сообщения kafka в POJO с использованием Spring Kafka. Я хочу использовать части ключа и значения сообщения для создания POJO.

Ключ сообщения kafka - это строка.
Значение сообщения kafka - JSON.

Я пытался сделать только часть значения сообщения, следуя инструкциям на codenotfound.com и baeldung.com . За исключением того, что я также хочу иметь значение ключа в POJO, и приложение java не генерирует сообщение.

Как получить приложение java для соответствующей десериализации сообщения kafka в POJO?

Например:

key = "test"

{
  "value1": "1st value"
  "value2": "2nd value"
}

Воспроизводимый пример того, что я пытаюсь найти, можно найти по адресу: https://github.com/gl3h/Simple-Consumer

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

Запустить команду docker-compose up -d, чтобы вызвать 3 экземпляра Zookeeper и Kafka. Он также вызывает Kafdrop, который подключается к кластеру Kafka.
Запустите приложение java. (gradle bootrun)

Отправка сообщения в данные topi c

kafka-console-producer --broker-list kafka1:29092 --topic data --property "parse.key=true" --property "key.separator=&" 
test&{"value1":"1st value","value2":"2nd value"}```

Всякий раз, когда сообщение отправляется в кластер Kafka, java приложению не удается преобразовать сообщение в объект POJO.

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.example.consumer.example.ExampleConsumer.processData(com.example.consumer.example.Data)]
Bean [com.example.consumer.example.ExampleConsumer@7a5a16cf]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}], failedMessage=GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}], failedMessage=GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.decorateException(KafkaMessageListenerContainer.java:1641) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeErrorHandler(KafkaMessageListenerContainer.java:1630) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1546) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeWithRecords(KafkaMessageListenerContainer.java:1487) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeRecordListener(KafkaMessageListenerContainer.java:1401) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeListener(KafkaMessageListenerContainer.java:1165) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.pollAndInvoke(KafkaMessageListenerContainer.java:949) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:884) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_221]
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_221]
    at java.lang.Thread.run(Thread.java:748) [na:1.8.0_221]
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}], failedMessage=GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:314) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:86) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:51) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1592) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.invokeOnMessage(KafkaMessageListenerContainer.java:1575) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeRecordListener(KafkaMessageListenerContainer.java:1534) [spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    ... 8 common frames omitted
Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.example.consumer.example.Data] for GenericMessage [payload={"value1":"value","value2":"value"}, headers={kafka_offset=1, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@2ff54c21, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=test, kafka_receivedPartitionId=0, kafka_receivedTopic=data, kafka_receivedTimestamp=1583036480453, kafka_groupId=data_consumer}]
    at org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver.resolveArgument(PayloadMethodArgumentResolver.java:145) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor$KafkaNullAwarePayloadArgumentResolver.resolveArgument(KafkaListenerAnnotationBeanPostProcessor.java:905) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite.resolveArgument(HandlerMethodArgumentResolverComposite.java:117) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.getMethodArgumentValues(InvocableHandlerMethod.java:148) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.messaging.handler.invocation.InvocableHandlerMethod.invoke(InvocableHandlerMethod.java:116) ~[spring-messaging-5.2.4.RELEASE.jar:5.2.4.RELEASE]
    at org.springframework.kafka.listener.adapter.HandlerAdapter.invoke(HandlerAdapter.java:48) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:304) ~[spring-kafka-2.3.0.RELEASE.jar:2.3.0.RELEASE]
    ... 13 common frames omitted

Нужно ли писать собственный десериализатор?
Как будет выглядеть пользовательский десериализатор?

groo · Answer 1 · 06 марта 2020

добро пожаловать в StackOverflow!

По умолчанию Spring Kafka использует десериализатор строк при потреблении сообщения, поэтому в вашем случае похоже, что вы хотите десериализовать сообщение Json, для этого первым шагом будет регистрация в качестве десериализатора значения для быть JsonDeserializer.class. Это должно работать для значений сообщения, но все еще не решает ключ, который вы также хотите.

В Kafka сериализаторы Key и Value не объединены, поэтому я не думаю, что есть простой способ получить ключ во время десериализации, возможно, простейшие варианты:

Создайте ключевую часть вашего Json объекта, чтобы он был автоматически десериализован с помощью JsonDeserliazer.
Процесс на стороне потребителя получает вместо самого объекта, но вместо этого использует ConsumerRecord, который вернет десериализованный ключ и значение, так что вы можете просто добавить ключ к десериализованному объекту, используя сеттер.

Надеюсь, это поможет уточнить. Я кратко рассмотрю ваш пример на Github и сделаю пиар, готово. Итак, чтобы исправить это, используя подход, чтобы ключ был частью полезной нагрузки сообщения (проверьте PR в своем репо):

Добавьте ключ к объекту данных как свойство и для своего потребителя:

 @Component
public class ExampleConsumer {

    @KafkaListener(topics = "data")
    public void processData(Data data) {
        System.out.println("Data:" + data);
    }
}

И добавьте правильную конфигурацию application.yml:

spring:
  application:
    name: kafka-consumer-example
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      group-id: data_consumer
      client-id: ${spring.application.name}
      properties:
        spring.json.value.default.type: com.example.consumer.example.Data
        spring.json.type.mapping: "data:com.example.consumer.example.Data"
        spring.json.trusted.packages: "*"

    listener:
      missing-topics-fatal: false

PS - Также для локального запуска kafka я бы рекомендовал использовать docker -compose файл с одним zookeeper и kafka, посмотрите этот пример -> https://dev.to/thegroo/spring-kafka-producer-and-consumer-41oc или этот другой -> https://dev.to/thegroo/one-to-run-them-all-1mg6

Gary Russell · Answer 2 · 06 марта 2020

Ваш слушатель не использует ваш собственный контейнерный завод; назовите его kafkaListenerContainerFactory для замены загрузки или используйте свойство containerFactory на слушателе.

@Bean
public ConcurrentKafkaListenerContainerFactory<String, Data> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Data> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(dataConsumerFactory());
    return factory;
}

и

Data:Data{key='null', value1='1st value', value2='2nd value'}

Или, что еще проще, настройте загрузку с помощью свойств.

Как десериализовать сообщение кафки в POJO?

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

Ответы [ 2 ]

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Как десериализовать сообщение кафки в POJO?

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

Ответы [ 2 ]

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Похожие темы