Confluent Platform与Logstash集成：Kafka消息编码异常排查

嘿，我来帮你拆解下这个问题：你看到的乱码，本质是Kafka主题里的消息用了Avro二进制序列化，但Logstash没做Avro解码，直接把原始字节输出了。

为什么会这样？ #

• 你用的是Confluent的JDBC源连接器，它默认会把从MySQL拉取的数据用Avro格式序列化，而且依赖Confluent Schema Registry来管理Avro的schema。这也是kafka-avro-console-consumer能正常输出结构化JSON的原因——这个消费者天生就支持Avro，会自动去Schema Registry拉取对应的schema来解码消息。
• 但你当前的Logstash Kafka输入配置，只是简单订阅了主题，没有告诉它要处理Avro格式。Logstash默认会把消息当作原始字节流处理，所以输出的就是Avro二进制数据转成字符后的乱码。

怎么解决？ #

要让Logstash正确解析Avro消息，得做两步：

1. 安装Logstash的Avro解码插件 #

首先给你的Logstash装个能解码Avro的插件，执行这条命令：

bin/logstash-plugin install logstash-codec-avro

2. 修改Logstash配置，开启Avro解码 #

在你的Kafka输入块里，添加Avro codec的配置，指定Schema Registry的地址（因为解码需要从这里拿schema）。修改后的配置如下：

kafka {
  bootstrap_servers => "localhost:9092"
  topics => ["requests_Operation"]
  add_field => { "[@metadata][flag]" => "operation" }
  # 新增Avro解码配置，替换成你的Schema Registry地址
  codec => avro {
    schema_registry_url => "http://localhost:8081"
  }
}

output {
  if [@metadata][flag] == "operation" {
    stdout { codec => rubydebug }
  }
}

3. 验证效果 #

重启Logstash后，你应该就能看到和kafka-avro-console-consumer一样的结构化JSON输出了。

额外注意点 #

如果你的Schema Registry开启了认证（比如用户名密码），还要在Avro codec里加上认证参数：

codec => avro {
  schema_registry_url => "http://localhost:8081"
  schema_registry_user => "你的用户名"
  schema_registry_password => "你的密码"
}

另外要确认下logstash-codec-avro插件和你的Logstash 6.2.4版本兼容，官方插件一般都会适配对应版本的Logstash。

内容的提问来源于stack exchange，提问作者Nikita Lipatov