The custom kafka data source not support json data map list format parser

[nimuyuhan]

I use below sql create source table:
create source table topic1(
uuid int,
data var
) with (
type = 'kafka',
kafka_topic = 'flow_message_test',
"auto.offset.reset" = latest,
kafka_broker = '10.16.98.10:9092',
kafka_group_id = 'test_json_parser',
value_type = 'json'
);
Data are as follows:
{
"uuid": 10455,
"data": ["{"applyid":"1590224987","base_info":"{\"city_id\":201}}"]
}

That sql grammar check pass. running throw exceptions blow:
java.lang.ClassCastException: java.util.ArrayList cannot be cast to java.lang.String
at org.apache.flink.api.common.typeutils.base.StringSerializer.copy(StringSerializer.java:33)
at org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:96)
at org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:47)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:577)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)
at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:718)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:696)
at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111)
at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordWithTimestamp(AbstractFetcher.java:398)
at org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher.emitRecord(Kafka010Fetcher.java:89)
at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:154)
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:738)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:94)
at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:58)
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704)
at java.lang.Thread.run(Thread.java:748)