如何配置Logstash Kafka并使用JsonDeserializer？配置报错求助

你遇到的错误是因为配置里多写了value_deserializer_class这一项，该类不在Logstash的类加载路径中，而且这个配置本身完全多余。

Logstash的Kafka输入插件已经封装了Kafka客户端的逻辑，你已经配置了codec => json {}，这个配置会自动处理JSON格式的消息解析，根本不需要手动指定Kafka原生的JsonDeserializer类。

修正后的配置 #

input {
  kafka{
    codec => json {}
    bootstrap_servers => "my_brokers"
    security_protocol => "SASL_PLAINTEXT"
    sasl_mechanism => "SCRAM-SHA-256"
    sasl_jaas_config => "my security config"
    topics => ["MY_TOPIC"]
  }
}

output {
  elasticsearch {
      hosts => ["localhost:9200"]
      index => "my_index"
  }
}

直接移除value_deserializer_class这一行即可解决问题，Logstash会通过你指定的codec来处理消息的反序列化，无需依赖Kafka原生的反序列化类。

内容的提问来源于stack exchange，提问作者dark ninja