kafka connect Transform base64解码字段

Question

我使用kafka RMQ源连接器从RMQ队列中获取数据。其中一个字段是base64编码的，它有一个json结构。我正在使用Extract变换来提取这个字段，但我不确定如何解码这个字段，我试图编写自己的smt来解码这个字段。但当我解码这个字段并试图将其放入记录的updatedvalue时，我遇到了错误java.lang.ClassCastException: [B cannot be cast to java.lang.String，因为消息是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主题的主要字段。如果我只是使用提取转换，那么它工作得很好，但我只能在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="
}

Answer 1

您需要使用ByteArrayConverter

"value.converter": "org.apache.kafka.connect.converters.ByteArrayConverter"

这里有一个例子。