看起来你的问题核心出在Kafka输入插件的配置过于简略，缺少必要参数来正确连接Kafka集群并消费指定主题的数据。我来帮你修正配置并梳理关键要点：

1. 修正Kafka输入配置 #

你原来的input { kafka { } }没有指定任何核心参数，Logstash根本不知道要连接哪个Kafka集群、消费哪个主题。补充以下关键参数后就能正常消费数据：

input {
  kafka {
    bootstrap_servers => "localhost:9092"  # 和你之前NetFlow转Kafka的配置一致，指向Kafka节点地址
    topic_id => "test"                     # 对应你之前输出到Kafka的主题名
    group_id => "logstash-es-consumer"     # 自定义消费组ID，避免重复消费或集群内冲突
    auto_offset_reset => "latest"          # 首次运行建议选"latest"（从最新消息开始消费），要拉取历史数据选"earliest"
    codec => "json"                        # 匹配Kafka中数据的编码格式，确保解析正常
  }
}

2. 优化Elasticsearch输出配置 #

原输出配置虽指定了hosts，但可以添加索引规则等参数，让ES的索引管理更规范（适配你用的Logstash 6.1.1版本）：

output {
  elasticsearch {
    hosts => ["120.127.XXX.XX:9200"]
    index => "netflow-%{+YYYY.MM.dd}"  # 按日期生成索引，方便后续查询和生命周期管理
    document_type => "_doc"            # 6.x版本ES必须指定该参数，7.x及以后版本无需配置
  }
  stdout { codec => rubydebug }        # 保留这个调试输出，方便你直观看到数据是否正常流转
}

3. 关键验证步骤 #

• 先确认Kafka主题有数据：用Kafka自带的命令行消费者验证test主题是否有NetFlow数据：

  kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
  

• 运行修正后的Logstash配置，观察控制台的rubydebug输出，如果能看到结构化的NetFlow数据，说明Kafka消费环节正常
• 访问ES的索引列表（比如浏览器打开http://120.127.XXX.XX:9200/_cat/indices?v），查看是否生成了netflow-YYYY.MM.dd格式的索引，且有数据写入
• 如果仍有问题，查看Logstash完整日志（路径D:/ELK/logstash-6.1.1/logs/logstash-plain.log），根据报错信息排查（比如网络不通、权限问题、版本兼容）

4. 版本兼容提醒 #

你用的是Logstash 6.1.1，要确保：

• Kafka版本在0.11.x~2.0.x之间，和Logstash 6.x系列兼容
• Elasticsearch版本尽量选6.x系列，避免跨版本API不兼容问题

内容的提问来源于stack exchange，提问作者張皓翔