kafkaedit

  • Version: 5.1.5
  • Released on: 2017-01-25
  • Changelog

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 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

Works with the old Ruby Event API (event['product']['price'] = 10)

0.9

2.4.x - 5.x.x

4.x.x

Works with the new getter/setter APIs (event.set('[product][price]', 10))

0.10.0.x

2.4.x - 5.x.x

5.x.x

Not compatible with the ⇐ 0.9 broker

Note

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.

This output supports connecting to Kafka over:

  • SSL (requires plugin version 3.0.0 or later)
  • Kerberos SASL (requires plugin version 5.1.0 or later)

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, 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}"
        }
        topic_id => "mytopic"
      }
    }
For more information see http://kafka.apache.org/documentation.html#theproducer

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

 

Synopsisedit

This plugin supports the following configuration options:

Required configuration options:

kafka {
    topic_id => ...
}

Available configuration options:

Setting Input typeRequiredDefault value

acks

string, one of ["0", "1", "all"]

No

"1"

batch_size

number

No

16384

bootstrap_servers

string

No

"localhost:9092"

buffer_memory

number

No

33554432

client_id

string

No

codec

codec

No

"plain"

compression_type

string, one of ["none", "gzip", "snappy", "lz4"]

No

"none"

enable_metric

boolean

No

true

id

string

No

jaas_path

a valid filesystem path

No

kerberos_config

a valid filesystem path

No

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

request_timeout_ms

string

No

retries

number

No

0

retry_backoff_ms

number

No

100

sasl_kerberos_service_name

string

No

sasl_mechanism

string

No

"GSSAPI"

security_protocol

string, one of ["PLAINTEXT", "SSL", "SASL_PLAINTEXT", "SASL_SSL"]

No

"PLAINTEXT"

send_buffer_bytes

number

No

131072

ssl_key_password

password

No

ssl_keystore_location

a valid filesystem path

No

ssl_keystore_password

password

No

ssl_keystore_type

string

No

ssl_truststore_location

a valid filesystem path

No

ssl_truststore_password

password

No

ssl_truststore_type

string

No

topic_id

string

Yes

value_serializer

string

No

"org.apache.kafka.common.serialization.StringSerializer"

workers

<<,>>

No

1

Detailsedit

 

acksedit

  • 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_sizeedit

  • 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 (DEPRECATED)edit

  • DEPRECATED WARNING: This configuration item is deprecated and may not be available in future versions.
  • 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_serversedit

  • 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_memoryedit

  • 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_idedit

  • 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

codecedit

  • Value type is codec
  • Default value is "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.

compression_typeedit

  • Value can be any of: none, gzip, snappy, lz4
  • 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.

enable_metricedit

  • Value type is boolean
  • Default value is 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 collection for a specific plugin.

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

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 useful when you have two or more plugins of the same type, for example, if you have 2 grok filters. Adding a named ID in this case will help in monitoring Logstash when using the monitoring APIs.

output {
 stdout {
   id => "my_plugin_id"
 }
}

jaas_pathedit

  • Value type is path
  • There is no default value for this setting.

The Java Authentication and Authorization Service (JAAS) API supplies user authentication and authorization services 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 these to the global JVM system properties. This means if you have multiple Kafka inputs, all of them would be sharing the same jaas_path and kerberos_config. If this is not desirable, you would have to run separate instances of Logstash on different JVM instances.

kerberos_configedit

  • Value type is path
  • There is no default value for this setting.

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

key_serializeredit

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

Serializer class for the key of the message

linger_msedit

  • 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_sizeedit

  • Value type is number
  • Default value is 1048576

The maximum size of a request

message_keyedit

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

The key for the message

metadata_fetch_timeout_msedit

  • Value type is number
  • Default value is 60000

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

metadata_max_age_msedit

  • Value type is number
  • Default value is 300000

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

receive_buffer_bytesedit

  • Value type is number
  • Default value is 32768

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

reconnect_backoff_msedit

  • 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.

request_timeout_msedit

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

The configuration controls the maximum amount of time the client will wait for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted.

retriesedit

  • 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_msedit

  • 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.

sasl_kerberos_service_nameedit

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

The Kerberos principal name that Kafka broker runs as. This can be defined either in Kafka’s JAAS config or in Kafka’s config.

sasl_mechanismedit

  • Value type is string
  • Default value is "GSSAPI"

SASL mechanism used for client connections. This may be any mechanism for which a security provider is available. GSSAPI is the default mechanism.

security_protocoledit

  • Value can be any of: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
  • Default value is "PLAINTEXT"

Security protocol to use, which can be either of PLAINTEXT,SSL,SASL_PLAINTEXT,SASL_SSL

send_buffer_bytesedit

  • Value type is number
  • Default value is 131072

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

ssl (DEPRECATED)edit

  • DEPRECATED WARNING: This configuration item is deprecated and may not be available in future versions.
  • Value type is boolean
  • Default value is false

Enable SSL/TLS secured communication to Kafka broker.

ssl_key_passwordedit

  • Value type is password
  • There is no default value for this setting.

The password of the private key in the key store file.

ssl_keystore_locationedit

  • Value type is path
  • There is no default value for this setting.

If client authentication is required, this setting stores the keystore path.

ssl_keystore_passwordedit

  • Value type is password
  • There is no default value for this setting.

If client authentication is required, this setting stores the keystore password

ssl_keystore_typeedit

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

The keystore type.

ssl_truststore_locationedit

  • Value type is path
  • There is no default value for this setting.

The JKS truststore path to validate the Kafka broker’s certificate.

ssl_truststore_passwordedit

  • Value type is password
  • There is no default value for this setting.

The truststore password

ssl_truststore_typeedit

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

The truststore type.

timeout_ms (DEPRECATED)edit

  • DEPRECATED WARNING: This configuration item is deprecated and may not be available in future versions.
  • 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_idedit

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

The topic to produce messages to

value_serializeredit

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

Serializer class for the value of the message

workersedit

  • Value type is string
  • Default value is 1

TODO remove this in Logstash 6.0 when we no longer support the :legacy type This is hacky, but it can only be herne