我已经用java创建了kafka streams项目。Kafka流停止并抛出如下错误:
eption:进程中捕获异常。taskid=0\u 18,processor=kstream-source-0000000000,topic=chil raw,partition=18,offset=414977位于org.apache.kafka.streams.processor.internals.streamtask.process(streamtask)。java:232)在org.apache.kafka.streams.processor.internals.assignedtasks.process(assignedtasks。java:403)在org.apache.kafka.streams.processor.internals.taskmanager.process(taskmanager。java:317)在org.apache.kafka.streams.processor.internals.streamthread.processandmaybecommit(streamthread。java:942)在org.apache.kafka.streams.processor.internals.streamthread.runonce(streamthread。java:822)在org.apache.kafka.streams.processor.internals.streamthread.runloop(streamthread。java:774)在org.apache.kafka.streams.processor.internals.streamthread.run(streamthread。java:744)原因:org.apache.kafka.common.errors.serializationexception:序列化avro消息时出错原因:java.lang.illegalstateexception:为locationgeofenceevent键创建的架构对象太多!在io.confluent.kafka.schemaregistry.client.cachedschemaregistryclient.register(cachedschemaregistryclient。java:152)位于io.confluent.kafka.serializers.abstractkafkaavroserializer.serializeimpl(abstractkafkaavroserializer)。java:79)在io.confluent.kafka.serializers.kafkaavroserializer.serialize(kafkaavroserializer。java:53)在io.confluent.kafka.streams.serdes.avro.genericavroserializer.serialize(genericavroserializer。java:63)at io.confluent.kafka.streams.serdes.avro.genericavroserializer.serialize(genericavroserializer。java:39)在org.apache.kafka.streams.processor.internals.recordcollectorimpl.send(recordcollectorimpl。java:90)在org.apache.kafka.streams.processor.internals.recordcollectorimpl.send(recordcollectorimpl。java:78)网址:org.apache.kafka.streams.processor.in
我使用genericavroserde和下面的代码导致了这个问题。请帮帮我
@Bean
public GenericAvroSerde genericValueSerde() {
GenericAvroSerde serde = new GenericAvroSerde();
serde.configure(Collections.singletonMap("schema.registry.url",
properties.getSchemaRegistryUrl()), false); // XXX `false` for record values
return serde;
}
暂无答案!
目前还没有任何答案,快来回答吧!