在Kibana、logstash add_字段中将Kafka producer ip/端口显示为字段?

我有ElasticSearch&基巴纳7.6.2 我将logstash与卡夫卡联系起来如下:

input {
 kafka {
   bootstrap_servers => "------"
   topics_pattern => ["----"]
   decorate_events => true   
 }
}
filter {
 mutate { add_field => { "[topic_name]" => "%{[@metadata][kafka][topic]}"} }
 mutate { add_field => { "[ip_port]" => "X*X*X*X*X*X*X"} }
 date   { match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ] }
}
output {
 elasticsearch {
  hosts => ["localhost:9200"]
  index => "logstash"
  document_type => "logs"
 }
}

我应该替换什么而不是 X*X*X*X*X*X*X 有消息的生产者IP/端口吗?
我指的是制作人将消息发布到卡夫卡时的客户端IP。