汇合Kafka流-找不到类IO.Confluent.Connect.Avro.ConnectDefault

Question

我使用的是一个带有查询模式的jdbc源连接器，似乎没有指定的表名，在schema-registry中注册的记录键和记录值的模式名称为空，并被分配默认名称“ConnectDefault”，如Confluent的AvroData类https://github.com/confluentinc/schema-registry/blob/master/avro-converter/src/main/Java/io/confluent/connect/avro/AvroData.Java中定义的

当使用生成的avro源和SpecificAvroSerde运行Kafka Streams应用程序时,我得到错误：

Exception in thread "streams-app-6e39ebfd-db14-49bc-834f-afaf108a6d25-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Failed to deserialize value for record. topic=topic-name, partition=0, offset=0
  at org.apache.kafka.streams.processor.internals.SourceNodeRecordDeserializer.deserialize(SourceNodeRecordDeserializer.java:46)
  at org.apache.kafka.streams.processor.internals.RecordQueue.addRawRecords(RecordQueue.java:84)
  at org.apache.kafka.streams.processor.internals.PartitionGroup.addRawRecords(PartitionGroup.java:117)
  at org.apache.kafka.streams.processor.internals.StreamTask.addRecords(StreamTask.java:474)
  at org.apache.kafka.streams.processor.internals.StreamThread.addRecordsToTasks(StreamThread.java:642)
  at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:548)
  at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:519)
Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing Avro message for id 2
Caused by: org.apache.kafka.common.errors.SerializationException: Could not find class io.confluent.connect.avro.ConnectDefault specified in writer's schema whilst finding reader's schema for a SpecificRecord.

我试图发布主题中的键模式和值模式的新版本，以表名作为模式名，并删除具有\“name\”：\“connectDefault\”、\“namespace\”：\“io.confluent.connect.avro\”属性的原始版本，但没有成功。我是缺少了一个名为ConnectDefault的类，还是可以在源连接器的某个地方指定一个没有命名空间的架构名称？

我的Kafka流配置：

streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-app");
streamsConfig.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
streamsConfig.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
streamsConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

我的Kafka Connect配置：

name=source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:oracle:thin:
mode=incrementing
incrementing.column.name=id
query=QUERY
topic.prefix=topic-name

transforms=InsertKey, ExtractId
transforms.InsertKey.type=org.apache.kafka.connect.transforms.ValueToKey
transforms.InsertKey.fields=id
transforms.ExtractId.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.ExtractId.field=id
key.converter=org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false

value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schemas.enable=true
value.converter.schema.registry.url=http://localhost:8081

Answer 1

问题是，当jdbc源连接器处于查询模式时，模式名默认为null。https://github.com/confluentinc/kafka-connect-jdbc/issues/90

看起来这可以通过使用SetSchemaMetadata转换在源连接器中添加带有SMT（Single Message Transforms）的模式名称来解决。https://cwiki.apache.org/confluence/display/kafka/kip-66%3a+single+message+transforms+for+kafka+connect

transforms=setValueSchema
transforms.setValueSchema.type=org.apache.kafka.connect.transforms.SetSchemaMetadata$Value
transforms.setValueSchema.schema.name=io.confluent.connect.avro.ConnectDefault