jdbc kafka连接avro模式与流不兼容

Question

我正在使用jdbc源代码连接器从oracle db获取数据，并按下（两个键

Value schema:
{
"subject": "testtopicA-value",
"version": 1,
"id": 1122,
"schema": "{\"type\":\"record\",\"name\":\"ConnectDefault\",\"namespace\":\"io.confluent.connect.avro\",\"fields\":[{\"name\":\"ID\",\"type\":[\"null\",\"long\"],\"default\":null},{\"name\":\"TIME\",\"type\":{\"type\":\"long\",\"connect.version\":1,\"connect.name\":\"org.apache.kafka.connect.data.Timestamp\",\"logicalType\":\"timestamp-millis\"}},{\"name\":\"STATUS\",\"type\":\"string\"},{\"name\":\"NUMBER\",\"type\":[\"null\",\"int\"],\"default\":null}]}"
}

Key Schema :
{
"subject": "testtopicA-key",
"version": 1,
"id": 1123,
"schema": "[\"null\",\"long\"]"
}

我有一个Kafka流收听这个主题，并有avro Genericrecord。当我启动流时，我开始得到＜code＞ClassCastException:java.lang.Long不能强制转换为org.apache.avro.generic。GenericRecordconnect生成的架构具有数据类型为“long”的字段

        final StreamsBuilder builder = new StreamsBuilder();
    KStream<?,?> stream;
    stream = builder.stream(INPUTTOPIC);        
    
    ((KStream<GenericRecord,GenericRecord>) stream)
    
    .filter((k,v) -> v != null)
    .map((k,v)->
    {
        .........
        .......
    

有人对如何解决这个问题有什么建议吗？融合版本:6.2.0

Answer 1

根据错误，您没有将默认键/值serde设置为Avro；你把它当成了朗塞德