我使用elasticsearch和kibana 7.6.2将logstash连接到kafka，如下所示：

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/端口？
我指的是制作人将消息发布到Kafka时的客户端ip。