当尝试将 record.value() 转换为 java 对象时,我在消费者中遇到了这个异常:
ClassCastException: class org.apache.avro.generic.GenericData$Record cannot be cast to class [...].PublicActivityRecord (org.apache.avro.generic.GenericData$Record and [...].PublicActivityRecord are in unnamed module of loader 'app')
生产者发送java对象,这是一个名为PublicActivityRecord
的用户定义类型,如下所示:
KafkaProducer<String, PublicActivityRecord> producer = new KafkaProducer<>(createKafkaProperties());
[...]
this.producer.send(new ProducerRecord<String, PublicActivityRecord>(myTopic, activityRecord));
this.producer.flush();
此时我可以在调试模式下看到ProducerRecord
的值确实是
PublicActivityRecord
类型。在注册服务器上,我可以在日志中看到生产者发送模式的 POST 请求:
Registering new schema: subject DEV-INF_9325_activityRecord_01-value, version null, id null, type null, schema size 7294 (io.confluent.kafka.schemaregistry.rest.resources.SubjectVersionsResource:262)
[2022-01-28 07:01:35,575] INFO 192.168.36.30 - - [28/janv./2022:06:01:34 +0000] "POST /subjects/DEV-INF_9325_activityRecord_01-value/versions HTTP/1.1" 200 8 "-" "Java/11.0.2" POSTsT (io.confluent.rest-utils.requests:62)
在消费者方面:
protected KafkaConsumer<String, PublicActivityRecord> consumer;
[...]
consumer = new KafkaConsumer<>(consumerProperties);
consumer.subscribe(Stream.of(kafkaConfig.getTopicActivityRecord()).collect(Collectors.toList()));
final ConsumerRecords<String, PublicActivityRecord> records = consumer.poll(duration);
records.forEach(record -> {
[...]
PublicActivityRecord activityRecord = record.value();
这里发生了ClassCastException。在调试模式下,我可以看到
record.value
确实是
GenericData$Record
类型。并且它不能转换为 PublicActivityRecord。序列化器/反序列化器的键和值是相同的:
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
在 schema-registry 日志中,我可以看到消费者的 GET 请求:
"GET /schemas/ids/3?fetchMaxId=false HTTP/1.1" 200 8447 "-" "Java/11.0.7" GETsT (io.confluent.rest-utils.requests:62)
所以我已经检查过:
PublicActivityRecord
GenericData$Record
所以问题是:为什么消费者得到的是
GenericData
记录而不是预期的
PublicActivityRecord
?任何线索将不胜感激!
value.deserializer.specific.avro.reader=true
或者,在您的消费者配置中使用常量
KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG
=真
.consumerProperty(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG,true)
示例:
@Bean
public ReceiverOptions<String, OrderRequestModel> receiverOptions() {
Map<String, Object> propetiesMap = new HashMap<>();
propetiesMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
propetiesMap.put("schema.registry.url", "http://localhost:8081");
propetiesMap.put(ConsumerConfig.GROUP_ID_CONFIG, group);
propetiesMap.put(ConsumerConfig.GROUP_INSTANCE_ID_CONFIG, "1");
propetiesMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
propetiesMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propetiesMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
return ReceiverOptions.<String, OrderRequestModel>create(propetiesMap)
.consumerProperty(JsonDeserializer.VALUE_DEFAULT_TYPE, OrderRequestModel.class)
.consumerProperty(JsonDeserializer.USE_TYPE_INFO_HEADERS, false)
.consumerProperty(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true)
.subscription(List.of(topic));
}