Использование темы kafka с помощью logstash для elasticSearch

Question

Я начинаю употреблять сообщения от kafka до logstash, и я хочу отправить всю тему в elasticSearch, но я не получаю никаких сообщений о logstash, я вижу в kafka сообщения приходят, но с стороны kafka я ничего не вижу, как правильно его настроить?

input { 
    kafka { 
    zk_connect => "localhost:2181" 
    topic_id => "event" 
    } 
} 

output{ 
    stdout{ 
    codec => rubydebug 
    } 
    elasticsearch{ 
    index => "event-%{+YYYY.MM.dd}" 
    hosts => ["localhost:9201"] 
    codec => json 
    } 
} 

curl localhost:9201 
{ 
    "name" : "Flex", 
    "cluster_name" : "elasticsearch", 
    "version" : { 
    "number" : "2.3.4", 
    "build_hash" : "e455fd0c13dceca8dbbdbb1665d068ae55dabe3f", 
    "build_timestamp" : "2016-06-30T11:24:31Z", 
    "build_snapshot" : false, 
    "lucene_version" : "5.5.0" 
    }, 
    "tagline" : "You Know, for Search" 
} 

команда:

/kafka-console-consumer.sh --zookeeper localhost:2181 --topic event 

дает результаты, время от времени.

Answer 1

Попробуйте, как это с auto_offset_reset и reset_beginning:

kafka { 
    topic_id => "event" 
    zk_connect => "localhost:2181" 
    group_id => "event-group" 
    auto_offset_reset => "smallest" 
    reset_beginning => true 
    consumer_threads => 1 
    }