How to parse message use ELK+Kafka

Recently I tried to analyzed the log with the stream:filebeat -> kafka ->logstash -> ES

The data stored in ES becomes the structure shown below:

{
   “_source”:{
           “@timestamp” :  "YYYY-MM-DDTHH:mm:ss.sss"
           "message":""" "@timestamp" :   "YYYY-MM-DDTHH:mm:ss.sss",  "message": "raw data"         """"  }
}

The "raw data" I actually want to parse is in message:{"message" : "raw data"}, which FileBeat didn't have to worry about when sending data directly to Logstash in the past, now that Kafka is added, what do I do?

Well, when I add "codec = >;Json ", the data is correct.....