Kafka input plugin

The frequency in milliseconds that the consumer offsets are committed to Kafka.

What to do when there is no initial offset in Kafka or if an offset is out of range:

A list of URLs of Kafka instances to use for establishing the initial connection to the cluster. This list should be in the form of host1:port1,host2:port2 These urls are just used for the initial connection to discover the full cluster membership (which may change dynamically) so this list need not contain the full set of servers (you may want more than one, though, in case a server is down).

Automatically check the CRC32 of the records consumed. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance.

How DNS lookups should be done. If set to use_all_dns_ips, when the lookup returns multiple IP addresses for a hostname, they will all be attempted to connect to before failing the connection. If the value is resolve_canonical_bootstrap_servers_only each entry will be resolved and expanded into a list of canonical names.

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.

A rack identifier for the Kafka consumer. Used to select the physically closest rack for the consumer to read from. The setting corresponds with Kafka’s broker.rack configuration.

Close idle connections after the number of milliseconds specified by this config.

Ideally you should have as many threads as the number of partitions for a perfect balance — more threads than partitions means that some threads will be idle

Option to add Kafka metadata like topic, message size and header key values to the event. This will add a field named kafka to the logstash event containing the following attributes:

  * Value type is boolean * Default value is true

Controls whether the topic is automatically created when subscribing to a non-existent topic. A topic will be auto-created only if this configuration is set to true and auto-topic creation is enabled on the broker using auto.create.topics.enable; otherwise auto-topic creation is not permitted. 

This committed offset will be used when the process fails as the position from which the consumption will begin.

If true, periodically commit to Kafka the offsets of messages already returned by the consumer. If value is false however, the offset is committed every time the consumer writes data fetched from the topic to the in-memory or persistent queue.

Whether records from internal topics (such as offsets) should be exposed to the consumer. If set to true the only way to receive records from an internal topic is subscribing to it.

The maximum amount of data the server should return for a fetch request. This is not an absolute maximum, if the first message in the first non-empty partition of the fetch is larger than this value, the message will still be returned to ensure that the consumer can make progress.

The maximum amount of time the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy fetch_min_bytes. This should be less than or equal to the timeout used in poll_timeout_ms

The minimum amount of data the server should return for a fetch request. If insufficient data is available the request will wait for that much data to accumulate before answering the request.

The identifier of the group this consumer belongs to. Consumer group is a single logical subscriber that happens to be made up of multiple processors. Messages in a topic will be distributed to all Logstash instances with the same group_id.

The static membership identifier for this Logstash Kafka consumer. Static membership feature was introduced in KIP-345, available under Kafka property group.instance.id. Its purpose is to avoid rebalances in situations in which a lot of data has to be forwarded after a consumer goes offline. This feature mitigates cases where the service state is heavy and the rebalance of one topic partition from instance A to B would cause a huge amount of data to be transferred. A client that goes offline/online frequently can avoid frequent and heavy rebalances by using this option.

The expected time between heartbeats to the consumer coordinator. Heartbeats are used to ensure that the consumer’s session stays active and to facilitate rebalancing when new consumers join or leave the group. The value must be set lower than session.timeout.ms, but typically should be set no higher than 1/3 of that value. It can be adjusted even lower to control the expected time for normal rebalances.

Controls how to read messages written transactionally. If set to read_committed, polling messages will only return transactional messages which have been committed. If set to read_uncommitted (the default), polling messages will return all messages, even transactional messages which have been aborted. Non-transactional messages will be returned unconditionally in either mode.

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.

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

Java Class used to deserialize the record’s key

The maximum amount of data per-partition the server will return. The maximum total memory used for a request will be #partitions * max.partition.fetch.bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition.

The maximum delay between invocations of poll() when using consumer group management. This places an upper bound on the amount of time that the consumer can be idle before fetching more records. If poll() is not called before expiration of this timeout, then the consumer is considered failed and the group will rebalance in order to reassign the partitions to another member.

The maximum number of records returned in a single call to poll().

The period of time in milliseconds after which we force a refresh of metadata even if we haven’t seen any partition leadership changes to proactively discover any new brokers or partitions

The name of the partition assignment strategy that the client uses to distribute partition ownership amongst consumer instances, supported options are:

These map to Kafka’s corresponding ConsumerPartitionAssignor implementations.

Time Kafka consumer will wait to receive new messages from topics.

After subscribing to a set of topics, the Kafka consumer automatically joins the group when polling. The plugin poll-ing in a loop ensures consumer liveness. Underneath the covers, Kafka client sends periodic heartbeats to the server. The timeout specified the time to block waiting for input on each poll.

The size of the TCP receive buffer (SO_RCVBUF) to use when reading data.

The amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly connecting to a host in a tight loop. This backoff applies to all requests sent by the consumer to the broker.

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.

The amount of time to wait before attempting to retry a failed fetch request to a given topic partition. This avoids repeated fetching-and-failing in a tight loop.

The SASL client callback handler class the specified SASL mechanism should use.

JAAS configuration setting local to this plugin instance, as opposed to settings using config file configured using jaas_path, which are shared across the JVM. This allows each plugin instance to have its own configuration.

If both sasl_jaas_config and jaas_path configurations are set, the setting here takes precedence.

Example (setting for Azure Event Hub):

    input {
      kafka {
        sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='auser'  password='apassword';"
      }
    }

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 mechanism used for client connections. This may be any mechanism for which a security provider is available. GSSAPI is the default mechanism.

Set the username for basic authorization to access remote Schema Registry.

Set the address of a forward HTTP proxy. An empty string is treated as if proxy was not set.

Set the password for basic authorization to access remote Schema Registry.

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

If schema registry authentication is required, this setting stores the keystore password.

The format of the keystore file. It must be either jks or PKCS12.

The truststore path to validate the schema registry’s certificate.

The schema registry truststore password.

The format of the schema registry’s truststore file. It must be either jks or PKCS12.

The URI that points to an instance of the Schema Registry service, used to manage Avro schemas. Be sure that the Avro schemas for deserializing the data from the specified topics have been uploaded to the Schema Registry service. The schemas must follow a naming convention with the pattern <topic name>-value.

Use either the Schema Registry config option or the value_deserializer_class config option, but not both.

When using the schema registry, by default the plugin checks connectivity and validates the schema registry, during plugin registration, before events are processed. In some circumstances, this process may fail when it tries to validate an authenticated schema registry, causing the plugin to crash. This setting allows the plugin to skip validation during registration, which allows the plugin to continue and events to be processed. Note that an incorrectly configured schema registry will still stop the plugin from processing events.

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

The size of the TCP send buffer (SO_SNDBUF) to use when sending data

The timeout after which, if the poll_timeout_ms is not invoked, the consumer is marked dead and a rebalance operation is triggered for the group identified by group_id

The endpoint identification algorithm, defaults to "https". Set to empty string "" to disable endpoint verification

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 format of the keystore file. It must be either jks or PKCS12.

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

The truststore password.

The format of the truststore file. It must be either jks or PKCS12.

A list of topics to subscribe to, defaults to ["logstash"].

A topic regular expression pattern to subscribe to.

Filtering by a regular expression is done by retrieving the full list of topic names from the broker and applying the pattern locally. When used with brokers with a lot of topics this operation could be very slow, especially if there are a lot of consumers.

Java Class used to deserialize the record’s value. A custom value deserializer can be used only if you are not using a Schema Registry. Use either the value_deserializer_class config option or the schema_registry_url config option, but not both.

Add a field to an event

The codec used for input data. Input codecs are a convenient method for decoding your data before it enters the input, without needing a separate filter in your Logstash pipeline.

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.

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 kafka inputs. Adding a named ID in this case will help in monitoring Logstash when using the monitoring APIs.

input {
  kafka {
    id => "my_plugin_id"
  }
}

Add any number of arbitrary tags to your event.

This can help with processing later.

Add a type field to all events handled by this input.

Types are used mainly for filter activation.

The type is stored as part of the event itself, so you can also use the type to search for it in Kibana.

If you try to set a type on an event that already has one (for example when you send an event from a shipper to an indexer) then a new input will not override the existing type. A type set at the shipper stays with that event for its life even when sent to another Logstash server.