Kafka Configuration

LogScale uses Apache Kafka internally for queuing incoming messages and for storing shared state when running LogScale in a cluster setup. This page describes how to configure Kafka.

LogScale has built-in API endpoints for controlling Kafka. Using the API it is possible to specify partition size and replication factor on the ingest queue. It's also possible to use other Kafka tools, such as the command-line tools included in the Kafka distribution.

Note

Make sure to not apply compression inside Kafka to the queues below. LogScale compresses the messages when relevant. Letting Kafka apply compression as well slows down the system and also adds problems with GC due to use of JNI in case LZ4 is applied. Setting compression.type to producer should be applied to these queues.

Topic Management

It is possible to use Kafka in two modes; either LogScale manages its Kafka topics (this is the default), or it does not. If LogScale is managing, it will create topics if they do not exist, and will look at the topic configurations and manage those as well. If LogScale is not managing the Kafka topics, it will not create topics or change configurations; you must create and properly configure the topics listed in the Topics section in Kafka.

To disable LogScale's automatic management, and manage topics manually, set the configuration flag KAFKA_MANAGED_BY_HUMIO to false.

Topic Configuration

We recommend that the ingest queue is configured with a backstop limit on how much data it will retain. During normal operation, LogScale will delete data from the ingest queue once it is no longer needed, but there are error scenarios

where this might not happen. The most common examples we have seen of this not happening include:

  • Bucket Storage is configured, and LogScale is configured for USING_EPHEMERAL_DISKS=true, and the bucket storage solution is experiencing downtime. In this configuration, logs are retained in Kafka until they can be replicated to bucket storage.

  • The LogScale cluster is partially down, such that ingest is still flowing into Kafka, but digest work is not being performed as the responsible nodes are not running. This can happen if most of the LogScale cluster is down, and may also happen in clusters making use of node roles if all nodes capable of digest work are down.

Since the result of data flowing into Kafka and not being deleted is that Kafka will fill its disk and eventually crash, which will pull down LogScale alongside it, we recommend setting a limit on retention in Kafka for the ingest queue such that the disk will never be completely filled.

We recommend setting the limit to be 90% of the disk size, divided by the partition count for the queue. For example, for a setup where Kafka is on a 1TB disk, and the LogScale ingest queue has 24 partitions, we would recommend setting the limit to 1TB / 24 = 41666666700 bytes.

The limit is configured using Kafka's management tooling. In your Kafka install directory, you should find a bin directory. Within that directory, run the command 

$ ./kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name QUEUE_NAME_HERE --add-config retention.bytes=YOUR_LIMIT_HERE

The queue name is determined by the KAFKA_QUEUE_PREFIX setting, which is empty by default. For example, assuming a 1TB disk and 24 partitions, and KAFKA_QUEUE_PREFIX=test-, you would run 

$ ./kafka-configs.sh --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name test-humio-ingest --add-config retention.bytes=41666666700

Kafka Client Properties

LogScale provides environment variables to add extra Kafka configuration properties to its Kafka consumers and Kafka producers. The environment variables have the following prefixes:

  • KAFKA_ADMIN_

  • KAFKA_CHATTER_CONSUMER_

  • KAFKA_CHATTER_PRODUCER_

  • KAFKA_GLOBAL_CONSUMER_

  • KAFKA_GLOBAL_PRODUCER_

  • KAFKA_INGEST_QUEUE_CONSUMER_

  • KAFKA_INGEST_QUEUE_PRODUCER_

In addition, KAFKA_COMMON_ can be utilized to pass the configuration to all clients; however, settings configured using the client-specific prefixes have precedence if the setting is present with both prefixes. Kafka configuration options, such as request.timeout.ms can be passed with these prefixes using a simple rewrite:

  1. Uppercase the option name. Example: REQUEST.TIMEOUT.MS

  2. Replace . with _. Example: REQUEST_TIMEOUT_MS

  3. Apply the prefix for the target client. Example: KAFKA_INGEST_QUEUE_CONSUMER_REQUEST_TIMEOUT_MS

  4. Pass this as an environment variable to LogScale on boot. Example: KAFKA_INGEST_QUEUE_CONSUMER_REQUEST_TIMEOUT_MS=30000

Retention Settings

The global configuration sets the following defaults:

  • retention.bytes - 20GB

  • retention.ms - -1 (minus 1, infinite)

Chatter sets:

  • retention.ms - to one hour

Ingest sets:

  • retention.ms - to 7 days, with no size limit

Important

Customers setting up production clusters should add a retention.bytes setting for the ingest queue, depending on how much disk space your brokers have, and what kind of ingest rate you expect. The limit will act as a backstop to prevent your Kafka brokers from running out of disk space, if LogScale isn't keeping up on consumption for a while. If you don't set a disk space limit, Kafka may fill its disks, and then you will likely need to do a Kafka reset to recover the cluster.

Sample Kafka Configuration

It is important to set log.dirs to the location where Kafka should store the data. Without such a setting, Kafka defaults to /tmp/kafka-logs, which is very likely NOT where you want it.

Important

The log.dirs parameter is the location of the Kafka data, not execution logs.

When deploying Kafka using ZooKeeper using the Tarball installation you need to configure the two services with a unique host ID number:

  • For Kafka this is the broker.id configuration value in the server.properties file:

    ini
    # The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

  • For ZooKeeper this is a file called myid in the ZooKeeper data directory that contains the node ID number. You can create it using:

    shell
    $ echo 1 >/kafka/zookeeper/myid

When creating a multi-node Kafka cluster these numbers must be unique for each host:

Host Kafka broker.id ZooKeeper myid
kafka1 1 1
kafka2 2 2
kafka3 3 3
ini
############################# Server Basics #############################

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1

############################# Socket Server Settings #############################

listeners=PLAINTEXT://kafka1:9092
#use compression
compression.type=producer

############################# Log Basics #############################

# A comma-separated list of directories under which to store log files
log.dirs=/kafka/kafka

############################# Log Retention Policy #############################

# The following configurations control the disposal of log segments. The policy can
# be set to delete segments after a period of time, or after a given size has accumulated.
# A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
# from the end of the log.

# The minimum age of a log file to be eligible for deletion
log.retention.hours=48

# A size-based retention policy for logs. Segments are pruned from the log as long as the remaining
# segments don't drop below log.retention.bytes.
#log.retention.bytes=1000073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000
auto.create.topics.enable=false
unclean.leader.election.enable=false

############################# ZooKeeper #############################
zookeeper.connect=kafka1:2181