Я пытаюсь десериализировать различные полезные нагрузки JSON из одной и той же темы Kafka. Другие вопросы, заданные здесь, направили меня к первой попытке, но я не смог запустить его. PrettyPrint-Override ">org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.foo.message.ConsumerImpl.consumeSelf(java.lang.String,java.lang.String,java.lang.String,java.lang.String,java.util.Map,com.foo.message.KafkaMessage,org.apache.kafka.clients.consumer.ConsumerRecord)]
Bean [com.foo.message.ConsumerImpl@6df2a206]; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: Cannot convert from [com.foo.message.KafkaMessageWithAdditionalField] to [com.foo.message.KafkaMessage] for GenericMessage [payload=com.foo.message.KafkaMessageWithAdditionalField@4e3168f7, headers={kafka_offset=22, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@c0e2fcf, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=fromBar, kafka_receivedTimestamp=1548310583481}], failedMessage=GenericMessage [payload=com.foo.message.KafkaMessageWithAdditionalField@4e3168f7, headers={kafka_offset=22, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@c0e2fcf, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=fromBar, kafka_receivedTimestamp=1548310583481}]
at org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.invokeHandler(MessagingMessageListenerAdapter.java:292) ~[spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:79) ~[spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
at org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter.onMessage(RecordMessagingMessageListenerAdapter.java:50) ~[spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.doInvokeOnMessage(KafkaMessageListenerContainer.java:1207) [spring-kafka-2.2.2.RELEASE.jar:2.2.2.RELEASE]
< /code>
...
loggingErrorHandler уже упоминает (правильное) значение в ConsumerRecord. < /p>
2019-01-24 07:16:27.630 ERROR 27204 --- [ntainer#2-0-C-1] o.s.kafka.listener.LoggingErrorHandler : Error while processing: ConsumerRecord(topic = fromBar, partition = 0, offset = 22, CreateTime = 1548310583481, serialized key size = -1, serialized value size = 196, headers = RecordHeaders(headers = [], isReadOnly = false), key = null, value = com.foo.bar.message.KafkaMessageWithAdditionalField@4e3168f7)
< /code>
Сначала моя конфигурация: < /p>
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public ConsumerFactory consumerFactoryMessage()
{
Map props = new HashMap();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
return new DefaultKafkaConsumerFactory(props, new StringDeserializer(),
new JsonDeserializer(KafkaMessage.class));
}
@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerMessageContainerFactory()
{
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactoryMessage());
return factory;
}
@Bean
public ConsumerFactory consumerFactoryMessageWithAdditionalField()
{
Map props = new HashMap();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
return new DefaultKafkaConsumerFactory(props, new StringDeserializer(),
new JsonDeserializer(KafkaMessageWithAdditionalField.class));
}
@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerMessageWithAdditionalFieldContainerFactory()
{
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setConsumerFactory(consumerFactoryMessageWithAdditionalField());
return factory;
}
}
< /code>
Вот слушатели: < /p>
@KafkaListener(topicPartitions = @TopicPartition(partitions = "0", topic = "${foo.kafka.topic-springBoot}"), containerFactory = "kafkaListenerMessageContainerFactory")
public void consumeSelf(@Headers Map map, KafkaMessage message, ConsumerRecord cr)
{
log.info("message received %s", message);
}
@KafkaListener(topicPartitions = @TopicPartition(partitions = "0", topic = "${foo.kafka.topic-springBoot}"), containerFactory = "kafkaListenerMessageWithAdditionalFieldContainerFactory")
public void consumeSelfAdd(@Headers Map map, KafkaMessageWithAdditionalField message, ConsumerRecord cr)
{
log.info("messageKafkaMessageWithAdditionalField received %s", message);
}
Подробнее здесь: https://stackoverflow.com/questions/543 ... ring-kafka
