IMPORTANT: No additional bug fixes or documentation updates will be released for this version. For the latest information, see the current release documentation.

« jira lumberjack »

› ›

kafka

edit

kafka

edit

Write events to a Kafka topic. This uses the Kafka Producer API to write messages to a topic on the broker.

Here’s a compatibility matrix that shows the Kafka client versions that are compatible with each combination of Logstash and the Kafka output plugin:

Kafka Client Version	Logstash Version	Plugin Version	Security Features	Why?
0.8	2.0.0 - 2.x.x	<3.0.0		Legacy, 0.8 is still popular
0.9	2.0.0 - 2.3.x	3.x.x	Basic Auth, SSL

We recommended that you use matching Kafka client and broker versions. During upgrades, you should upgrade brokers before clients because brokers target backwards compatibility. For example, the 0.9 broker is compatible with both the 0.8 consumer and 0.9 consumer APIs, but not the other way around.

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:

    output {
      kafka {
        codec => plain {
           format => "%{message}"
        }
      }
    }
For more information see http://kafka.apache.org/documentation.html#theproducer

Kafka producer configuration: http://kafka.apache.org/documentation.html#newproducerconfigs

Synopsis

edit

This plugin supports the following configuration options:

Required configuration options:

kafka {
    topic_id => ...
}

Available configuration options:

Setting	Input type	Required	Default value
`acks`	string, one of `["0", "1", "all"]`	No	`"1"`
`batch_size`	number	No	`16384`
`block_on_buffer_full`	boolean	No	`true`
`bootstrap_servers`	string	No	`"localhost:9092"`
`buffer_memory`	number	No	`33554432`
`client_id`	string	No
`codec`	codec	No	`"json"`
`compression_type`	string, one of `["none", "gzip", "snappy"]`	No	`"none"`
`key_serializer`	string	No	`"org.apache.kafka.common.serialization.StringSerializer"`
`linger_ms`	number	No	`0`
`max_request_size`	number	No	`1048576`
`message_key`	string	No
`metadata_fetch_timeout_ms`	number	No	`60000`
`metadata_max_age_ms`	number	No	`300000`
`receive_buffer_bytes`	number	No	`32768`
`reconnect_backoff_ms`	number	No	`10`
`retries`	number	No	`0`
`retry_backoff_ms`	number	No	`100`
`send_buffer_bytes`	number	No	`131072`
`timeout_ms`	number	No	`30000`
`topic_id`	string	Yes
`value_serializer`	string	No	`"org.apache.kafka.common.serialization.StringSerializer"`
`workers`	number	No	`1`

Details

edit

`acks`

edit

Value can be any of: 0, 1, all
Default value is "1"

The number of acknowledgments the producer requires the leader to have received before considering a request complete.

acks=0, the producer will not wait for any acknowledgment from the server at all. acks=1, This will mean the leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. acks=all, This means the leader will wait for the full set of in-sync replicas to acknowledge the record.

`batch_size`

edit

Value type is number
Default value is 16384

The producer will attempt to batch records together into fewer requests whenever multiple records are being sent to the same partition. This helps performance on both the client and the server. This configuration controls the default batch size in bytes.

`block_on_buffer_full`

edit

Value type is boolean
Default value is true

When our memory buffer is exhausted we must either stop accepting new records (block) or throw errors. By default this setting is true and we block, however in some scenarios blocking is not desirable and it is better to immediately give an error.

`bootstrap_servers`

edit

Value type is string
Default value is "localhost:9092"

This is for bootstrapping and the producer will only use it for getting metadata (topics, partitions and replicas). The socket connections for sending the actual data will be established based on the broker information returned in the metadata. The format is host1:port1,host2:port2, and the list can be a subset of brokers or a VIP pointing to a subset of brokers.

`buffer_memory`

edit

Value type is number
Default value is 33554432

The total bytes of memory the producer can use to buffer records waiting to be sent to the server.

`client_id`

edit

Value type is string
There is no default value for this setting.

The id string to pass to the server when making requests. The purpose of this is to be able to track the source of requests beyond just ip/port by allowing a logical application name to be included with the request

`codec`

edit

Value type is codec
Default value is "json"

The codec used for output data. Output codecs are a convenient method for encoding your data before it leaves the output, without needing a separate filter in your Logstash pipeline.

`compression_type`

edit

Value can be any of: none, gzip, snappy
Default value is "none"

The compression type for all data generated by the producer. The default is none (i.e. no compression). Valid values are none, gzip, or snappy.

`key_serializer`

edit

Value type is string
Default value is "org.apache.kafka.common.serialization.StringSerializer"

Serializer class for the key of the message

`linger_ms`

edit

Value type is number
Default value is 0

The producer groups together any records that arrive in between request transmissions into a single batched request. Normally this occurs only under load when records arrive faster than they can be sent out. However in some circumstances the client may want to reduce the number of requests even under moderate load. This setting accomplishes this by adding a small amount of artificial delay—that is, rather than immediately sending out a record the producer will wait for up to the given delay to allow other records to be sent so that the sends can be batched together.

`max_request_size`

edit

Value type is number
Default value is 1048576

The maximum size of a request

`message_key`

edit

Value type is string
There is no default value for this setting.

The key for the message

`metadata_fetch_timeout_ms`

edit

Value type is number
Default value is 60000

the timeout setting for initial metadata request to fetch topic metadata.

`metadata_max_age_ms`

edit

Value type is number
Default value is 300000

the max time in milliseconds before a metadata refresh is forced.

`receive_buffer_bytes`

edit

Value type is number
Default value is 32768

The size of the TCP receive buffer to use when reading data

`reconnect_backoff_ms`

edit

Value type is number
Default value is 10

The amount of time to wait before attempting to reconnect to a given host when a connection fails.

`retries`

edit

Value type is number
Default value is 0

Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error.

`retry_backoff_ms`

edit

Value type is number
Default value is 100

The amount of time to wait before attempting to retry a failed produce request to a given topic partition.

`send_buffer_bytes`

edit

Value type is number
Default value is 131072

The size of the TCP send buffer to use when sending data.

`timeout_ms`

edit

Value type is number
Default value is 30000

The configuration controls the maximum amount of time the server will wait for acknowledgments from followers to meet the acknowledgment requirements the producer has specified with the acks configuration. If the requested number of acknowledgments are not met when the timeout elapses an error will be returned. This timeout is measured on the server side and does not include the network latency of the request.

`topic_id`

edit

This is a required setting.
Value type is string
There is no default value for this setting.

The topic to produce messages to

`value_serializer`

edit

Value type is string
Default value is "org.apache.kafka.common.serialization.StringSerializer"

Serializer class for the value of the message

`workers`

edit

Value type is number
Default value is 1

The number of workers to use for this output. Note that this setting may not be useful for all outputs.

« jira lumberjack »