logstash input多个kafka异常

问题描述

graph LR;
    filebeat --> logstash;
    log4j --> logstash;
    logstash --> es;

filebeat 和 log4j appender 同时到 kafka, logstash在启动时报错, 错误如下:

javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=logstash-0

问题原因及解决

input 消费kafka时, 分别指定不同的 client_id.

kafka {
        bootstrap_servers => ["192.168.103.43:9092"] # 注意这里配置的kafka的broker地址不是zk的地址
        client_id => "kafka_client_1"
        group_id => "logstash"
        topics => ["ipaynow_log"]  # kafka topic 名称
        consumer_threads => 5
        decorate_events => true
        type => "string"
        codec => "json"
}

kafka {
        bootstrap_servers => ["192.168.103.43:9092"] # 注意这里配置的kafka的broker地址不是zk的地址
        client_id => "kafka_client_2"
        group_id => "logstash"
        topics => ["ipaynow-hunter"]  # kafka topic 名称
        consumer_threads => 5
        decorate_events => true
        type => "string"
        codec => plain { charset=>"UTF-8" }
}