For other versions, see theVersioned plugin docs.
For questions about the plugin, open a topic in the Discuss forums. For bugs or feature requests, open an issue in Github.For the list of Elastic supported plugins, please consult the Elastic Support Matrix.
Write events to a Kafka topic.
This plugin uses Kafka Client 2.1.0. For broker compatibility, see the official Kafka compatibility reference. If the linked compatibility wiki is not up-to-date, please contact Kafka support/community to confirm compatibility.
If you require features not yet available in this plugin (including client version upgrades), please file an issue with details about what you need.
This output supports connecting to Kafka over:
By default security is disabled but can be turned on as needed.
The only required configuration is the topic_id.
The default codec is plain. Logstash will encode your events with not only the message field but also with a timestamp and hostname.
If you want the full content of your events to be sent as json, you should set the codec in the output configuration like this:
output { kafka { codec => json topic_id => "mytopic" }}
For more information see http://kafka.apache.org/documentation.html#theproducer
Kafka producer configuration: http://kafka.apache.org/documentation.html#newproducerconfigs
This plugin supports the following configuration options plus the Common Options described later.
Setting | Input type | Required |
---|---|---|
string, one of |
No |
|
No |
||
No |
||
No |
||
No |
||
string, one of |
No |
|
a valid filesystem path |
No |
|
a valid filesystem path |
No |
|
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
No |
||
string, one of |
No |
|
No |
||
No |
||
No |
||
a valid filesystem path |
No |
|
No |
||
No |
||
a valid filesystem path |
No |
|
No |
||
No |
||
Yes |
||
No |
Also see Common Options for a list of options supported by alloutput plugins.
0
, 1
, all
"1"
The number of acknowledgments the producer requires the leader to have receivedbefore 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.
16384
The producer will attempt to batch records together into fewer requests whenever multiplerecords are being sent to the same partition. This helps performance on both the clientand the server. This configuration controls the default batch size in bytes.
"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 beestablished based on the broker information returned in the metadata. The format ishost1:port1,host2:port2
, and the list can be a subset of brokers or a VIP pointing to asubset of brokers.
33554432
The total bytes of memory the producer can use to buffer records waiting to be sent to the server.
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 justip/port by allowing a logical application name to be included with the request
none
, gzip
, snappy
, lz4
"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.
The Java Authentication and Authorization Service (JAAS) API supplies user authentication and authorizationservices for Kafka. This setting provides the path to the JAAS file. Sample JAAS file for Kafka client:
KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useTicketCache=true renewTicket=true serviceName="kafka"; };
Please note that specifying jaas_path
and kerberos_config
in the config file will add theseto the global JVM system properties. This means if you have multiple Kafka inputs, all of them would be sharing the samejaas_path
and kerberos_config
. If this is not desirable, you would have to run separate instances of Logstash ondifferent JVM instances.
Optional path to kerberos config file. This is krb5.conf style as detailed in https://web.mit.edu/kerberos/krb5-1.12/doc/admin/conf_files/krb5_conf.html
"org.apache.kafka.common.serialization.StringSerializer"
Serializer class for the key of the message
0
The producer groups together any records that arrive in between requesttransmissions into a single batched request. Normally this occurs only underload when records arrive faster than they can be sent out. However in some circumstancesthe 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 delayto allow other records to be sent so that the sends can be batched together.
The key for the message
60000
the timeout setting for initial metadata request to fetch topic metadata.
300000
the max time in milliseconds before a metadata refresh is forced.
32768
The size of the TCP receive buffer to use when reading data
10
The amount of time to wait before attempting to reconnect to a given host when a connection fails.
The configuration controls the maximum amount of time the client will waitfor the response of a request. If the response is not received before the timeoutelapses the client will resend the request if necessary or fail the request ifretries are exhausted.
The default retry behavior is to retry until successful. To prevent data loss,the use of this setting is discouraged.
If you choose to set retries
, a value greater than zero will cause theclient to only retry a fixed number of times. This will result in data lossif a transport fault exists for longer than your retry count (network outage,Kafka down, etc).
A value less than zero is a configuration error.
100
The amount of time to wait before attempting to retry a failed produce request to a given topic partition.
The Kerberos principal name that Kafka broker runs as.This can be defined either in Kafka’s JAAS config or in Kafka’s config.
"GSSAPI"
SASL mechanism used for client connections.This may be any mechanism for which a security provider is available.GSSAPI is the default mechanism.
PLAINTEXT
, SSL
, SASL_PLAINTEXT
, SASL_SSL
"PLAINTEXT"
Security protocol to use, which can be either of PLAINTEXT,SSL,SASL_PLAINTEXT,SASL_SSL
131072
The size of the TCP send buffer to use when sending data.
"https"
The endpoint identification algorithm, defaults to "https"
. Set to empty string ""
to disable
The password of the private key in the key store file.
If client authentication is required, this setting stores the keystore path.
If client authentication is required, this setting stores the keystore password
The keystore type.
The JKS truststore path to validate the Kafka broker’s certificate.
The truststore password
The truststore type.
The topic to produce messages to
"org.apache.kafka.common.serialization.StringSerializer"
Serializer class for the value of the message
The following configuration options are supported by all output plugins:
"plain"
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.
true
Disable or enable metric logging for this specific plugin instance.By default we record all the metrics we can, but you can disable metrics collectionfor a specific plugin.
Add a unique ID
to the plugin configuration. If no ID is specified, Logstash will generate one.It is strongly recommended to set this ID in your configuration. This is particularly usefulwhen you have two or more plugins of the same type. For example, if you have 2 kafka outputs.Adding a named ID in this case will help in monitoring Logstash when using the monitoring APIs.
output { kafka { id => "my_plugin_id" }}