问题描述
我正在尝试反序列化来自同一Kafka主题的不同JSON负载.在这里提出的其他问题使我第一次尝试,但是我无法使其运行.
I'm trying to deserialize different JSON payloads from the same Kafka topic. The other questions asked here, guided me to a first attempt, but I was not able to get it running.
正如Gary提到的那样(这里),这里有一些提示( JsonSerializer.ADD_TYPE_INFO_HEADERS ),但是当我同时收发两条消息,但都出现异常.
As Gary mentioned (here) there is some hint (JsonSerializer.ADD_TYPE_INFO_HEADERS), but when I send and receive both messages I get an exception.
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<java.lang.String, java.lang.Object>,com.foo.message.KafkaMessage,org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.Object>)]
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]
...
LoggingErrorHandler在ConsumerRecord中已经提到了一个(正确的)值.
The LoggingErrorHandler mentions already a (the correct) value in ConsumerRecord.
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)
首先我的配置:
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
@Value("${spring.kafka.bootstrap-servers}")
private String bootstrapServers;
@Bean
public ConsumerFactory<String, KafkaMessage> consumerFactoryMessage()
{
Map<String, Object> 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<String, KafkaMessage> kafkaListenerMessageContainerFactory()
{
ConcurrentKafkaListenerContainerFactory<String, KafkaMessage> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactoryMessage());
return factory;
}
@Bean
public ConsumerFactory<String, KafkaMessageWithAdditionalField> consumerFactoryMessageWithAdditionalField()
{
Map<String, Object> 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<String, KafkaMessageWithAdditionalField> kafkaListenerMessageWithAdditionalFieldContainerFactory()
{
ConcurrentKafkaListenerContainerFactory<String, KafkaMessageWithAdditionalField> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactoryMessageWithAdditionalField());
return factory;
}
}
以下是听众:
@KafkaListener(topicPartitions = @TopicPartition(partitions = "0", topic = "${foo.kafka.topic-springBoot}"), containerFactory = "kafkaListenerMessageContainerFactory")
public void consumeSelf(@Headers Map<String, Object> map, KafkaMessage message, ConsumerRecord<String, Object> cr)
{
log.info("message received %s", message);
}
@KafkaListener(topicPartitions = @TopicPartition(partitions = "0", topic = "${foo.kafka.topic-springBoot}"), containerFactory = "kafkaListenerMessageWithAdditionalFieldContainerFactory")
public void consumeSelfAdd(@Headers Map<String, Object> map, KafkaMessageWithAdditionalField message, ConsumerRecord<String, Object> cr)
{
log.info("messageKafkaMessageWithAdditionalField received %s", message);
}
推荐答案
您不能这样做;您有2个不同的侦听器容器,其中的侦听器期望使用不同的对象.
You can't do that; you have 2 different listener containers with listeners that expect different objects.
对于接收不同类型的多个侦听器方法,您需要在类级别使用@KafkaListener
,在方法级别使用@KafkaHandler
.
For multiple listener methods that receive different types, you need to use @KafkaListener
at the class level and @KafkaHandler
at the method level.
请参见 @KafkaListener关于类.
@KafkaListener(id = "multi", topics = "myTopic")
static class MultiListenerBean {
@KafkaHandler
public void listen(String foo) {
...
}
@KafkaHandler
public void listen(Integer bar) {
...
}
@KafkaHandler(isDefault = true`)
public void listenDefault(Object object) {
...
}
}
默认方法是可选的,用于未知负载类型.
The default method is optional and is used for unknown payload types.
但这仅适用于智能解串器(知道如何转换为不同的有效负载).
But this only works with a smart deserializer (that knows how to convert to different payloads).
或者,您可以将RecordFilterStrategy
添加到侦听器容器工厂中,以跳过每个侦听器中的其他记录.
Or, you can add a RecordFilterStrategy
to the listener container factory to skip the other records in each listener.
这篇关于使用Spring Kafka反序列化来自同一Kafka主题的不同JSON有效负载的文章就介绍到这了,希望我们推荐的答案对大家有所帮助,也希望大家多多支持!