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.

LogScale also uses the Kafka Admin API to help set up Kafka initially. For more information, see Kafka Admin API.

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.

Kafka Usage

LogScale uses Apache Kafka internally for queueing incoming messages and for storing shared state when running LogScale in a cluster setup. This page describes how LogScale uses Kafka. If you already understand Kafka concepts, you can skip this and go to the instructions on how to install Kafka using containers, or on bare metal.

For more information on Kafka configuration and settings, see Kafka Configuration.

How LogScale Uses Kafka

LogScale creates the following queues in Kafka:

You can set the environment variable HUMIO_KAFKA_TOPIC_PREFIX to add that prefix to the topic names in Kafka. Adding a prefix is recommended if you share the Kafka installation with applications other than LogScale, or with another LogScale instance. The default is not to add a prefix.

LogScale configures default retention settings on the topics when it creates them. If they exist already, LogScale does not alter retention settings on the topics.

If you wish to inspect and change the topic configurations, such as the retention settings, to match your disk space available for Kafka, please use the kafka-configs command. See below for an example, modifying the retention on the ingest queue to keep burst of data for up to one hour only.

global-events

This is LogScale's event-sourced database queue.

  • This queue has a relatively low throughput.

  • Allow messages of at least 2MB or more to allow large events:

    ini
    max.message.bytes=2097152

  • No log data is saved to this queue.

  • There should be a high number of replicas for this queue.

  • LogScale will raise the number of replicas on this queue to three if there are at least three brokers in the Kafka cluster and LogScale is allowed to manage the topic.

Default required replicas:

ini
min.insync.replicas = 2

Provided there are three brokers when LogScale creates the topic. Default retention configuration:

ini
retention.bytes = 1073741824

Which configures 1GB, and disable time based retention:

ini
retention.ms = -1

Compression should be set to:

ini
compression.type=producer
kafka-humio-ingest

Ingested events are sent to this queue, before they are stored in LogScale. LogScale's front end will accept ingest requests, parse them, and put them on the queue. LogScale's back end processes events from the queue and stores them into the datastore. This queue will have high throughput corresponding to the ingest load. The number of replicas can be configured in accordance with data size, latency and throughput requirements, and how important it is not to lose in-flight data.

LogScale defaults to two replicas on this queue, if at least two brokers exist in the Kafka cluster, and LogScale has not been told otherwise through the configuration parameter INGEST_QUEUE_REPLICATION_FACTOR, which defaults to 2. When data is stored in LogScale's own datastore, we don't need it on the queue any more.

  • Default required replicas:

    ini
    min.insync.replicas = $INGEST_QUEUE_REPLICATION_FACTOR - 1

    Provided there are enough brokers when LogScale creates the topic.

  • Default retention configuration (7 days as milliseconds):

    ini
    retention.ms = 604800000

  • Set the retention configuration on the humio-ingesttopic to:

    ini
    retention.bytes = disk_space_in_bytes_on_one_host / partitionCount

    with the actual setting based on the disk space available.

  • Compression should be set to:

    ini
    compression.type=producer

  • Allow messages of at least 8 MB to allow large events:

    ini
    max.message.bytes=8388608

  • Compaction is not allowed.

transientChatter-events

This queue is used for chatter between LogScale nodes. It is only used for transient data. LogScale will raise the number of replicas on this queue to 3 if there are at least three brokers in the Kafka cluster. The queue can have a short retention and it is not important to keep the data, as it gets stale very fast.

  • Default required replicas (provided there are three brokers when LogScale creates the topic):

    ini
    min.insync.replicas = 2

  • Default retention configuration (one hour as millis):

    ini
    retention.ms = 3600000

  • Compression should be set to:

    ini
    compression.type=producer

  • Support compaction settings allowing Kafka to retain only the latest copy:

    ini
    cleanup.policy=delete,compact

Kafka Version

LogScale recommends that the latest Kafka version is used with your LogScale deployment. The latest version of Kafka is available at Kafka Downloads.

Note

The minimum supported Kafka version is Kafka 2.4.1 and greater and is usually tested against the latest Kafka version.

You can set the configuration for individual topics using the following commands:

shell Syntax
## Example commands for setting protocol version on topic...
# See current config for topic, if any:
kafka-configs.sh --zookeeper localhost:2181 --describe --entity-type topics --entity-name 'humio-ingest'
# Set protocol version for topic:
kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name 'humio-ingest' --add-config 'message.format.version=0.11.0'
# Remove setting, allowing to use the default of the broker:
kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name 'humio-ingest' --delete-config 'message.format.version'

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

CrowdStrike recommends 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 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.

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

LogScale requires Kafka 2.4.1 or later, deployed using KRaft mode (Kafka Raft). ZooKeeper-based Kafka deployments are no longer supported as ZooKeeper mode was removed in Kafka 4.0.

Important

The log.dirs parameter specifies the location of the Kafka data, not execution logs. Without this setting, Kafka defaults to /tmp/kafka-logs, which is probably not where you want it.

KRaft Mode Overview

In KRaft mode, Kafka uses an internal Raft-based consensus mechanism, eliminating the need for a separate ZooKeeper ensemble. Each Kafka node is assigned a unique node.id and a process.roles value that defines whether it acts as a broker, a controller, or both (broker,controller).

For production deployments, it is recommended to separate broker and controller roles across dedicated nodes. For smaller or single-node deployments, combined roles are acceptable.

Node ID assignment

Each node in the cluster must have a unique node.id. For a three-node cluster, assign IDs as follows:

Host node.id process.roles
kafka1 1 broker,controller
kafka2 2 broker,controller
kafka3 3 broker,controller

Generating a Cluster ID

Before starting a KRaft cluster for the first time, you must generate a cluster UUID and format the storage directories on each node.

Generate a cluster ID (run once):

shell
$ ./kafka-storage.sh random-uuid

Format storage on each node using the generated UUID:

shell
$ ./kafka-storage.sh format -t YOUR_CLUSTER_UUID_HERE -c /path/to/server.properties

Sample server.properties:

ini
############################# KRaft Mode Settings #############################

# The role(s) this node plays in the cluster.
# Options: broker, controller, or broker,controller
process.roles=broker,controller

# The unique ID for this node. Must be unique across all nodes in the cluster.
node.id=1

# The controller quorum voters list.
# Format: node.id@host:port
# List all controller nodes here, on every node in the cluster.
controller.quorum.voters=1@kafka1:9093,2@kafka2:9093,3@kafka3:9093

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

# Listeners for broker and controller communication.
listeners=PLAINTEXT://kafka1:9092,CONTROLLER://kafka1:9093

# Listener used for inter-broker communication.
inter.broker.listener.name=PLAINTEXT

# Listener used for controller communication.
controller.listener.names=CONTROLLER

# Listener security protocol map.
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT

############################# Compression #############################

# Do not apply additional compression. LogScale handles compression internally.
compression.type=producer

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

# The directory where Kafka will store its data.
# Important: This is data storage, not execution logs.
log.dirs=/kafka/kafka

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

# Retain log segments for a minimum of 48 hours.
log.retention.hours=48

# Uncomment and set this to limit disk usage per partition.
# See the Topic Configuration section for guidance on sizing.
#log.retention.bytes=41666666700

# How frequently to check log segments for deletion eligibility.
log.retention.check.interval.ms=300000

# Prevent automatic topic creation; LogScale manages its own topics.
auto.create.topics.enable=false

# Do not allow unclean leader elections to prevent potential data loss.
unclean.leader.election.enable=false

Note

The controller.quorum.voters list must be identical on every node in the cluster and must include all nodes assigned the controller role.

Migration from ZooKeeper to KRaft

If you are operating an existing LogScale deployment that uses a ZooKeeper-based Kafka cluster, migration to KRaft mode is required. Refer to the Apache Kafka KRaft Migration Guide for detailed migration steps. It is strongly recommended to plan and test this migration in a non-production environment before applying it to production clusters.