Class: LogStash::Outputs::Kafka
- Inherits:
-
Base
- Object
- Base
- LogStash::Outputs::Kafka
- Defined in:
- lib/logstash/outputs/kafka.rb
Overview
Write events to a Kafka topic. This uses the Kafka Producer API to write messages to a topic on the broker.
The only required configuration is the topic name. The default codec is json, so events will be persisted on the broker in json format. If you select a codec of plain, Logstash will encode your messages with not only the message but also with a timestamp and hostname. If you do not want anything but your message passing through, you should make the output configuration something like:
- source,ruby
-
output
kafka { codec => plain { format => "%{message" } topic_id => "mytopic" }}
For more information see kafka.apache.org/documentation.html#theproducer
Kafka producer configuration: kafka.apache.org/documentation.html#newproducerconfigs
Instance Method Summary collapse
- #close ⇒ Object
-
#receive(event) ⇒ Object
def register.
- #register ⇒ Object
Instance Method Details
#close ⇒ Object
146 147 148 |
# File 'lib/logstash/outputs/kafka.rb', line 146 def close @producer.close end |
#receive(event) ⇒ Object
def register
139 140 141 142 143 144 |
# File 'lib/logstash/outputs/kafka.rb', line 139 def receive(event) if event == LogStash::SHUTDOWN return end @codec.encode(event) end |
#register ⇒ Object
119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 |
# File 'lib/logstash/outputs/kafka.rb', line 119 def register @producer = create_producer @codec.on_event do |event, data| begin if @message_key.nil? record = org.apache.kafka.clients.producer.ProducerRecord.new(event.sprintf(@topic_id), data) else record = org.apache.kafka.clients.producer.ProducerRecord.new(event.sprintf(@topic_id), event.sprintf(@message_key), data) end @producer.send(record) rescue LogStash::ShutdownSignal @logger.info('Kafka producer got shutdown signal') rescue => e @logger.warn('kafka producer threw exception, restarting', :exception => e) end end end |