我正在使用kafka rmq源连接器从rmq队列获取数据。其中一个字段是base64encoded,它具有json结构。我使用提取转换提取这个字段,但我不知道如何解码这个字段,我试图写我自己的smt解码这个字段,但我得到的错误 java.lang.ClassCastException: [B cannot be cast to java.lang.String
当我解码此字段并尝试将其放入记录的updatedvalue时,因为消息是一个json结构。非常感谢你的帮助。
下面是我的连接器配置。
{
"name" : "RabbitMQ_Source",
"connector.class" : "com.github.jcustenborder.kafka.connect.rabbitmq.RabbitMQSourceConnector",
"tasks.max" : "1",
"kafka.topic" : "RMQ_Topic",
"rabbitmq.queue" : "rmqqueue",
"rabbitmq.username":"username",
"rabbitmq.virtual.host":"dummy",
"rabbitmq.password":"password",
"rabbitmq.host":"x.x.x.x",
"rabbitmq.port":"5674",
"transforms": "ExtractField",
"transforms.ExtractField.type":"org.apache.kafka.connect.transforms.ExtractField$Value",
"transforms.ExtractField.field":"body"
}
下面是我试图使用的队列中的消息,body是解码后需要发送给kafka topic的主字段。如果我只是使用提取转换,那么它工作得很好,但我只能看到Kafka主题中的编码消息。
{
"consumerTag": "abcd",
"envelope": {
"deliveryTag": 1,
"isRedeliver": false,
"exchange": "rmqqueue",
"routingKey": "rmqqueue"
},
"basicProperties": {
"contentType": "text/plain",
"contentEncoding": null,
"headers": {},
"deliveryMode": 2,
"priority": 0,
"correlationId": null,
"replyTo": null,
"expiration": null,
"messageId": null,
"timestamp": null,
"type": null,
"userId": null,
"appId": null
},
"body": "eyJXSFMiOlt7IkNoYXJhY3RlciBTZXQiOiJVVEYtOCIsImFjdGlvbiI6InJld3JpdGUiLCJVcGRhdGUtRGF0ZS1UaW1lIjoiMjAyMC0wMy0yNSAwOTowMDowMjoxOSJ9XX0="
}
1条答案
按热度按时间btqmn9zl1#
你需要使用
ByteArrayConverter
```"value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter"