Is needed to convert MessageExt body to String in WorkerSinkTask convertToSinkDataEntry method.

```java

private SinkDataEntry convertToSinkDataEntry(MessageExt message) {
Map properties = message.getProperties();
String queueName;
EntryType entryType;
Schema schema;
Long timestamp;
Object[] datas = new Object[1];
if (null == recordConverter || recordConverter instanceof RocketMQConverter) {
queueName = properties.get(RuntimeConfigDefine.CONNECT_TOPICNAME);
String connectEntryType = properties.get(RuntimeConfigDefine.CONNECT_ENTRYTYPE);
entryType = StringUtils.isNotEmpty(connectEntryType) ? EntryType.valueOf(connectEntryType) : null;
String connectTimestamp = properties.get(RuntimeConfigDefine.CONNECT_TIMESTAMP);
timestamp = StringUtils.isNotEmpty(connectTimestamp) ? Long.valueOf(connectTimestamp) : null;
String connectSchema = properties.get(RuntimeConfigDefine.CONNECT_SCHEMA);
schema = StringUtils.isNotEmpty(connectSchema) ? JSON.parseObject(connectSchema, Schema.class) : null;
datas = new Object[1];
datas[0] = new String(message.getBody());
} else {
final byte[] messageBody = message.getBody();
final SourceDataEntry sourceDataEntry = JSON.parseObject(new String(messageBody), SourceDataEntry.class);
final Object[] payload = sourceDataEntry.getPayload();
final byte[] decodeBytes = Base64.getDecoder().decode((String) payload[0]);
Object recodeObject;
if (recordConverter instanceof JsonConverter) {
JsonConverter jsonConverter = (JsonConverter) recordConverter;
jsonConverter.setClazz(Object[].class);
recodeObject = recordConverter.byteToObject(decodeBytes);
datas = (Object[]) recodeObject;
}
schema = sourceDataEntry.getSchema();
entryType = sourceDataEntry.getEntryType();
queueName = sourceDataEntry.getQueueName();
timestamp = sourceDataEntry.getTimestamp();
}
DataEntryBuilder dataEntryBuilder = new DataEntryBuilder(schema);
dataEntryBuilder.entryType(entryType);
dataEntryBuilder.queue(queueName);
dataEntryBuilder.timestamp(timestamp);
List fields = schema.getFields();
if (null != fields && !fields.isEmpty()) {
for (Field field : fields) {
dataEntryBuilder.putFiled(field.getName(), datas[field.getIndex()]);
}
}
SinkDataEntry sinkDataEntry = dataEntryBuilder.buildSinkDataEntry(message.getQueueOffset());
return sinkDataEntry;
}
```
datas[0] = new String(message.getBody());
IMO, it will occur some problem in encode and decode.

Start by locating WorkerSinkTask.convertToSinkDataEntry and inspect the new String(message.getBody()) conversion in the default converter branch. Reproduce the reported encode/decode problem with a MessageExt body and verify that the resulting SinkDataEntry preserves the original message data.

Written by the indexing model from the issue text.