mutate add_field 在aggregate 的task_id 拿不到值

参见以下我的配置,aggregate 的task_id 显示的是 字符串“messageId”，不是对应的值<BA340DEBBECE2A4085C91373869DF0B686E546D3@shex-d03>。但是flag拿到的值是对的。不明白为什么 task_id => "%{messageId}" 不能取到实际值？烦请知道的解答。谢谢 。
filter {

mutate {
split => [ "message", ","]

add_field => {
"dateTime" => "%{[message][0]}"
"clientIp" => "%{[message][1]}"
"clientHost" => "%{[message][2]}"
"serverIp" => "%{[message][3]}"
"serverHost" => "%{[message][4]}"
"eventId" => "%{[message][8]}"
"messageId" => "%{[message][10]}"
"recipientAddress" => "%{[message][11]}"
"recipientStatus" => "%{[message][12]}"
"recipientCount" => "%{[message][14]}"
"relatedRecipientAddress" => "%{[message][15]}"
"senderAddress" => "%{[message][18]}"
"messageSubject" => "%{[message][17]}"

}
}
if [eventId] == "SEND" {
mutate {
add_field => {
"flag" => "%{messageId}"
}
}
aggregate {
task_id => "%{messageId}"
code => "map['test'] ||='a';map['test'] +='b';event.set('test2', map['test']);"
end_of_task => true
timeout => 20
# task_id => messageId
# code => "map[task_id] ||= ''; map[task_id] += %{messageId};"
# push_map_as_event_on_timeout => true
# timeout_task_id_field => "messageId"
# timeout => 20 # 1 hour timeout, user activity will be considered finished one hour after the first event, even if events keep coming
# inactivity_timeout => 20 # 5 minutes timeout, user activity will be considered finished if no new events arrive 5 minutes after the last event
# timeout_tags => ['_aggregatetimeout']
# timeout_code => "event.set('messageId_he', task_id)"
}

}else{
drop { }
}

}