如何通过Logstash消费Kafka消息？Kafka输入插件故障排查

核心问题回顾 #

• 环境：Docker Compose部署apache/kafka:3.9.0 + logstash:8.16.1
• 现象：Logstash可正常连接Kafka Broker，generator/http输入转Kafka输出功能正常；但Kafka输入转Kafka输出路径异常，报UNKNOWN_TOPIC_OR_PARTITION错误，手动创建logstash-input主题后问题依旧

1. 检查Logstash Kafka输入配置细节 #

重点核对logstash.conf中Kafka输入块的关键参数：

• 主题名称：确认topics参数与手动创建的主题完全一致（Kafka主题大小写敏感，无多余空格）
• Bootstrap地址：必须使用Docker Compose内部服务名（如kafka:9092），而非宿主机IP或localhost，否则Logstash容器无法正确解析Kafka节点
• 消费者组ID：必须配置group_id，Kafka消费者依赖该标识管理偏移量，缺失会引发异常
• 偏移量重置策略：建议设置auto_offset_reset => "earliest"，避免因无初始偏移量导致消费失败

示例正确配置片段：

input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["logstash-input"]
    group_id => "logstash-consumer-group"
    auto_offset_reset => "earliest"
    codec => "json"
  }
}

2. 验证Kafka主题与网络配置 #

主题有效性验证 #

在Kafka容器内执行命令，确认主题存在且配置正确：

# 查看所有主题
docker exec -it <kafka-container-name> kafka-topics.sh --list --bootstrap-server localhost:9092
# 查看目标主题详情
docker exec -it <kafka-container-name> kafka-topics.sh --describe --topic logstash-input --bootstrap-server localhost:9092

确保主题的分区数、副本数符合预期，无异常状态。

Kafka监听地址配置检查 #

检查docker-compose.yaml中Kafka服务的KAFKA_ADVERTISED_LISTENERS参数，必须包含容器内部可访问的地址，示例正确配置：

services:
  kafka:
    image: apache/kafka:3.9.0
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,PLAINTEXT_HOST://0.0.0.0:9093
      # 其他配置...

缺失PLAINTEXT://kafka:9092会导致Logstash容器无法获取正确的Kafka元数据。

3. 排查消费者偏移量与权限 #

偏移量状态检查 #

查看Logstash消费者组的偏移量情况：

docker exec -it <kafka-container-name> kafka-consumer-groups.sh --describe --group logstash-consumer-group --bootstrap-server localhost:9092

如果偏移量不存在或超出主题有效范围，执行偏移量重置：

docker exec -it <kafka-container-name> kafka-consumer-groups.sh --reset-offsets --to-earliest --topic logstash-input --group logstash-consumer-group --execute --bootstrap-server localhost:9092

ACL权限验证 #

如果Kafka开启了ACL，需确保Logstash的消费者组拥有logstash-input主题的消费权限，可通过以下命令添加：

docker exec -it <kafka-container-name> kafka-acls.sh --add --allow-principal User:* --operation Read --topic logstash-input --group logstash-consumer-group --bootstrap-server localhost:9092

4. 插件兼容性验证 #

检查Logstash Kafka输入插件版本与Kafka 3.9.0的兼容性：

docker exec -it <logstash-container-name> bin/logstash-plugin list --verbose logstash-input-kafka

若版本不匹配，更新插件：

docker exec -it <logstash-container-name> bin/logstash-plugin update logstash-input-kafka

5. 测试验证流程 #

1. 在Kafka容器内生产测试消息：

docker exec -it <kafka-container-name> kafka-console-producer.sh --topic logstash-input --bootstrap-server localhost:9092

输入测试内容，如{"test": "kafka-input-test"}

2. 查看Logstash容器日志，确认是否有消费记录
3. 检查Kafka输出目标主题，验证消息是否正常转发

内容的提问来源于stack exchange，提问作者1Z10