如何在logstash中检查来源是kafka还是beat？

input {
  kafka {
    bootstrap_servers => "kafka_server:9092"
    topics => ["your_topic"]
  }
}

input {
  beats {
    port => "5044"
  }
}

filter {
  if [source] =~ "kafka" {
    # 处理kafka来源的消息
  } else if [source] =~ "beat" {
    # 处理beat来源的消息
  }
}

output {
  if [source] =~ "kafka" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "kafka_index"
    }
  } else if [source] =~ "beat" {
    elasticsearch {
      hosts => ["localhost:9200"]
      index => "beat_index"
    }
  }
}