🥇Determining the appropriate size for a Kafka cluster in Kubernetes

Note: translation.: In this article, Banzai Cloud shares an example of using its special utilities to facilitate the management of Kafka within Kubernetes. The provided instructions illustrate how to determine the optimal size of infrastructure and configure Kafka itself to achieve the required throughput.

Apache Kafka is a distributed streaming platform for building reliable, scalable, and high-performance real-time streaming systems. Its impressive capabilities can be extended with Kubernetes. To this end, we have developed an Open Source Kafka operator and a tool called Supertubes. They allow you to run Kafka in Kubernetes and utilize its various features, such as fine-tuning broker configuration, metric-based scaling with rebalancing, rack awareness, graceful (graceful) rolling updates, etc.

Try Supertubes in your cluster:
curl https://getsupertubes.sh | sh and supertubes install -a --no-democluster --kubeconfig
Or consult the documentation. You can also read about some of the Kafka features that are automated using Supertubes and the Kafka operator. We have already written about them in our blog:

Oh no! Yet another Kafka operator for Kubernetes;

Monitor and operate Kafka based on Prometheus metrics;

Kafka rack awareness on Kubernetes;

Running Apache Kafka over Istio — benchmark;

User authenticated and access controlled clusters with the Kafka operator;

Kafka rolling upgrade and dynamic configuration on Kubernetes;

Envoy protocol filter for Kafka, meshed.

When deciding to deploy a Kafka cluster in Kubernetes, you will likely encounter the challenge of determining the optimal size of the underlying infrastructure and the need to fine-tune the Kafka configuration to meet throughput requirements. The maximum performance of each broker is determined by the performance of the underlying infrastructure components, such as memory, CPU, disk speed, network bandwidth, etc.

Ideally, the broker configuration should be such that all infrastructure elements are utilized to their maximum potential. However, in real life, such a setup is quite complex. It is more likely that users will configure broker settings to maximize the usage of one or two components (disk, memory, or processor). Generally speaking, a broker demonstrates its peak performance when its configuration allows the slowest component to be fully engaged. This way, we can get an approximate understanding of the load that a single broker can handle.

In theory, we can also estimate the number of brokers needed to handle a given load. However, in practice, the configuration options at various levels are so numerous that assessing the potential performance of a certain setup is quite challenging (if not impossible). In other words, it is very difficult to plan a configuration based on a given performance metric.

For Supertubes users, we typically apply the following approach: we start with a certain configuration (infrastructure + settings), then measure its performance, adjust the broker settings, and repeat the process again. This continues until the potential of the slowest component of the infrastructure is fully utilized.

In this way, we gain a clearer understanding of how many brokers the cluster needs to manage a specific load (the number of brokers also depends on other factors, such as the minimum number of message replicas for resilience, the number of partition leaders, and so on). Additionally, we get insights into which infrastructure component would benefit from vertical scaling.

This article will discuss the steps we take to 'squeeze everything' out of the slowest components in initial configurations and measure the throughput of the Kafka cluster. A highly resilient configuration requires at least three operational brokers (min.insync.replicas=3), distributed across three different availability zones. To configure, scale, and monitor the Kubernetes infrastructure, we use our own container management platform for hybrid clouds — Pipeline. It supports on-premise (bare metal, VMware) and five types of clouds (Alibaba, AWS, Azure, Google, Oracle), as well as any combinations of them.

Thoughts on the infrastructure and configuration of the Kafka cluster

For the examples provided below, we chose AWS as the cloud service provider and EKS as the Kubernetes distribution. A similar configuration can be implemented using PKE — a Kubernetes distribution from Banzai Cloud, certified by CNCF.

Disk

Amazon offers various types of EBS volumes. At the core of gp2 and io1 are SSDs, however, to ensure high throughput, gp2 it consumes accrued credits (I/O credits), which is why we preferred the type io1, that offers stable high throughput.

Instance types

Kafka performance is heavily dependent on the operating system's page cache, so we need instances with sufficient memory for brokers (JVM) and page cache. The instance c5.2xlarge is a decent start since it has 16 GB of memory and is optimized for EBS. Its drawback is that it can provide maximum performance for no more than 30 minutes every 24 hours. If the workload requires maximum performance for a longer period, other instance types should be considered. That's exactly what we did, settling on c5.4xlarge. It provides maximum throughput of 593.75 MB/s. The maximum throughput of an EBS volume io1 is higher than that of the instance c5.4xlarge, so the slowest component of the infrastructure is presumably the I/O throughput of this instance type (which should also be confirmed by the results of our load tests).

Network

Network bandwidth should be sufficiently large compared to the performance of the VM instance and disk; otherwise, the network becomes a bottleneck. In our case, the network interface c5.4xlarge supports speeds up to 10 Gb/s, which is significantly higher than the I/O throughput of the VM instance.

Deployment of brokers

Brokers should be deployed (planned in Kubernetes) on dedicated nodes to avoid competing with other processes for CPU, memory, network, and disk resources.

Java Version

Java 11 is a logical choice as it is compatible with Docker in that the JVM correctly identifies the processors and memory available to the container in which the broker runs. Knowing that CPU limits are important, the JVM internally and transparently sets the number of GC threads and JIT compiler threads. We used the Kafka image banzaicloud/kafka:2.13-2.4.0, which includes Kafka version 2.4.0 (Scala 2.13) on Java 11.

If you want to learn more about Java/JVM on Kubernetes, check out the following publications:

Why my Java application is OOMKilled;

How to correctly size containers for Java 10 applications.

Broker memory settings

There are two key aspects in configuring broker memory: settings for the JVM and for the Kubernetes pod. The memory limit set for the pod should be higher than the maximum heap size so that the JVM has room for Java metaspace, which exists in its own memory, and for the operating system's page cache, which Kafka actively uses. In our tests, we ran Kafka brokers with the parameters -Xmx4G -Xms2G, while the memory limit for the pod was 10 Gi. Note that JVM memory settings can be obtained automatically using -XX:MaxRAMPercentage and -X:MinRAMPercentage, based on the memory limit for the pod.

Broker CPU settings

In general, performance can be improved by increasing parallelism through an increase in the number of threads used by Kafka. The more processors available to Kafka, the better. In our test, we started with a limit of 6 processors and gradually (in iterations) increased that number to 15. Additionally, we set num.network.threads=12 in the broker settings to increase the number of threads handling incoming network data and sending it out. Immediately noticing that follower brokers could not fetch replicas quickly enough, we raised num.replica.fetchers to 4 to increase the speed at which follower brokers replicated messages from leaders.

Load generation tool

It is important to ensure that the load generator's capacity does not run out before the Kafka cluster (which is being benchmarked) reaches its maximum load. In other words, a preliminary assessment of the load generation tool's capabilities must be conducted, as well as the selection of instance types with sufficient processor and memory capacity. In this case, our tool will generate more load than the Kafka cluster can handle. After numerous experiments, we settled on three instances c5.4xlarge, each of which ran the generator.

Benchmarking

Performance measurement is an iterative process that includes the following stages:

setting up the infrastructure (EKS cluster, Kafka cluster, load generation tool, as well as Prometheus and Grafana);
generating load over a specified period to filter out random fluctuations in the collected performance metrics;
adjusting the infrastructure and broker configuration based on observed performance metrics;
repeating the process until the required throughput level of the Kafka cluster is achieved. During this, it must be consistently reproducible and demonstrate minimal variations in throughput.

The following section describes the steps performed during the benchmarking of the test cluster.

Tools

To quickly deploy the basic configuration, generate load, and measure performance, the following tools were used:

Banzai Cloud Pipeline to organize the EKS cluster from Amazon with Prometheus (for collecting Kafka and infrastructure metrics) and Grafana (for visualizing these metrics). We utilized integrated downward API support (simultaneously with this in Pipeline services that provide federated monitoring, centralized log collection, vulnerability scanning, disaster recovery, enterprise-grade security, and much more.
Sangrenel is a load testing tool for the Kafka cluster.
Grafana panels for visualizing Kafka and infrastructure metrics: Kubernetes Kafka, Node Exporter.
Supertubes CLI for easily configuring a Kafka cluster in Kubernetes. Zookeeper, Kafka operator, Envoy, and many other components are installed and properly configured to run a production-ready Kafka cluster in Kubernetes.
- To install supertubes CLI follow the instructions provided here.

EKS Cluster

Prepare the EKS cluster with dedicated worker nodes c5.4xlarge in different availability zones for pods with Kafka brokers, as well as dedicated nodes for the load generator and monitoring infrastructure.

banzai cluster create -f https://raw.githubusercontent.com/banzaicloud/kafka-operator/master/docs/benchmarks/infrastructure/cluster_eks_202001.json

Once the EKS cluster is up and running, enable its integrated monitoring service — it will deploy Prometheus and Grafana into the cluster.

Kafka System Components

Install Kafka system components (Zookeeper, kafka-operator) in EKS using the supertubes CLI:

supertubes install -a --no-democluster --kubeconfig

Kafka Cluster

By default, EKS uses EBS volumes of type gp2, so a separate storage class based on volumes needs to be created io1 for the Kafka cluster:

kubectl create -f - <<EOF
apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: fast-ssd
provisioner: kubernetes.io/aws-ebs
parameters:
  type: io1
  iopsPerGB: "50"
  fsType: ext4
volumeBindingMode: WaitForFirstConsumer
EOF

Set the parameter for brokers min.insync.replicas=3 and deploy broker pods on nodes in three different availability zones:

supertubes cluster create -n kafka --kubeconfig  -f https://raw.githubusercontent.com/banzaicloud/kafka-operator/master/docs/benchmarks/infrastructure/kafka_202001_3brokers.yaml --wait --timeout 600

Topics

We ran three instances of the load generator in parallel. Each writes to its topic, so we need a total of three topics:

supertubes cluster topic create -n kafka --kubeconfig  -f -<<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
  name: perftest1
spec:
  name: perftest1
  partitions: 12
  replicationFactor: 3
  retention.ms: '28800000'
  cleanup.policy: delete
EOF

supertubes cluster topic create -n kafka --kubeconfig  -f -<<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
    name: perftest2
spec:
  name: perftest2
  partitions: 12
  replicationFactor: 3
  retention.ms: '28800000'
  cleanup.policy: delete
EOF

supertubes cluster topic create -n kafka --kubeconfig  -f -<<EOF
apiVersion: kafka.banzaicloud.io/v1alpha1
kind: KafkaTopic
metadata:
  name: perftest3
spec:
  name: perftest3
  partitions: 12
  replicationFactor: 3
  retention.ms: '28800000'
  cleanup.policy: delete
EOF

For each topic, the replication factor is 3 — the minimum recommended value for high-availability production systems.

Load generation tool

We launched three instances of the load generator (each writing to a separate topic). For the load generator pods, it is necessary to specify node affinity so that they are scheduled only on the nodes assigned to them:

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  labels:
    app: loadtest
  name: perf-load1
  namespace: kafka
spec:
  progressDeadlineSeconds: 600
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: loadtest
  strategy:
    rollingUpdate:
      maxSurge: 25%
      maxUnavailable: 25%
    type: RollingUpdate
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: loadtest
    spec:
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
            - matchExpressions:
              - key: nodepool.banzaicloud.io/name
                operator: In
                values:
                - loadgen
      containers:
      - args:
        - -brokers=kafka-0:29092,kafka-1:29092,kafka-2:29092,kafka-3:29092
        - -topic=perftest1
        - -required-acks=all
        - -message-size=512
        - -workers=20
        image: banzaicloud/perfload:0.1.0-blog
        imagePullPolicy: Always
        name: sangrenel
        resources:
          limits:
            cpu: 2
            memory: 1Gi
          requests:
            cpu: 2
            memory: 1Gi
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30

Several points to consider:

The load generator generates messages of 512 bytes each and publishes them to Kafka in batches of 500 messages.
With the argument -required-acks=all publication is considered successful when all synchronized replicas of the message are received and acknowledged by the Kafka brokers. This means that in the benchmark we measured not only the speed of the leaders receiving messages, but also their followers replicating the messages. The task of this test does not include evaluating the read speed of consumers (consumers) of recently received messages that remain in the OS page cache, and comparing it with the read speed of messages stored on disk.
The load generator runs 20 workers in parallel (-workers=20). Each worker contains 5 producers, which jointly utilize the worker's connection to the Kafka cluster. As a result, each generator comprises 100 producers, all of whom send messages to the Kafka cluster.

Monitoring the status of the cluster

During the load testing of the Kafka cluster, we also monitored its health to ensure there were no pod restarts, desynchronized replicas, and maximum throughput with minimal fluctuations:

The load generator records standard statistics on the number of published messages and the error rate. The error rate should remain at 0,00%.
Cruise Control, deployed by the kafka-operator, provides a monitoring dashboard where we can also observe the cluster's state. To view this dashboard, execute:
```
supertubes cluster cruisecontrol show -n kafka --kubeconfig
```
ISR level (number of "in-sync" replicas) shrink and expansion equals 0.

Measurement results

3 brokers, message size — 512 bytes

With partitions evenly distributed across three brokers, we achieved a throughput of ~500 Mb/s (approximately 990 thousand messages per second):

The JVM virtual machine's memory consumption did not exceed 2 GB:

Disk throughput reached the maximum I/O capacity of the node on all three instances where the brokers were running:

From the node memory usage data, it appears that system buffering and caching took ~10-15 GB:

3 brokers, message size — 100 bytes

With a reduction in message size, throughput decreases by approximately 15-20%: this is affected by the time spent processing each message. Additionally, CPU load nearly doubled.

Since there are still unused cores on the broker nodes, performance can be improved by adjusting Kafka's configuration. This is a tricky task, so to increase throughput, it's better to work with larger messages.

4 brokers, message size — 512 bytes

Cluster performance can easily be increased by simply adding new brokers and maintaining the balance of partitions (which ensures an even distribution of load among brokers). In our case, after adding a broker, the cluster's throughput increased to ~580 Mb/s (~1.1 million messages per second). The increase was less than expected, primarily due to the imbalance of partitions (not all brokers are operating at peak capacity).

The memory usage of the JVM machine remained below 2 GB:

The work of brokers with the storages was affected by a partition imbalance:

Conclusions

The iterative approach presented above can be extended to cover more complex scenarios involving hundreds of consumers, repartitioning, rolling updates, pod restarts, etc. This allows us to evaluate the limits of Kafka cluster capabilities under various conditions, identify bottlenecks in its operation, and find ways to address them.

We developed Supertubes for fast and easy deployment of the cluster, its configuration, adding/removing brokers and topics, responding to alerts, and ensuring the proper functioning of Kafka in Kubernetes overall. Our goal is to help you focus on the main task of 'producing' and 'consuming' Kafka messages, while all the heavy lifting is handled by Supertubes and the Kafka operator.

If you are interested in the technologies and open-source projects by Banzai Cloud, follow the company on GitHub, LinkedIn or Twitter.