Self-Balancing Clusters¶

Important

Starting in Confluent Platform 6.0.0, Self-Balancing is offered as a preferred alternative to the Auto Data Balancer. For a detailed feature comparison, see Self-Balancing vs. Auto Data Balancer.
Self-Balancing and Auto Data Balancer cannot be used together. If you want to run Auto Data Balancer, you must first make sure that Self-Balancing is off.
Self-Balancing is a dynamic option, so you can turn it off or on while the cluster is running and set to rebalance when brokers are added, or for any uneven load (Anytime).

What are Self-Balancing Clusters?¶

Confluent Platform deployments can run hundreds of brokers, manage thousands of topics and produce billions of messages per hour. Every day brokers die, new topics are created and deleted, and partitions must be reassigned to balance the workload. This can overload teams tasked with managing Confluent Platform runtime operations.

Self-Balancing automates your resource workload balancing, provides failure detection and self-healing, and allows you to add or decommission brokers as needed, with no manual tuning required.

Self-Balancing offers:

Fully automated load balancing
Auto-monitoring of clusters for imbalances based on a large set of parameters, configurations, and runtime variables
Continuous metrics aggregation and rebalancing plans, generated instantaneously in most cases, and executed automatically
Automatic triggering of rebalance operations based on simple configurations you set on Confluent Control Center or in Kafka server.properties files. You can choose to auto-balance Only when brokers are added or Anytime, which rebalances for any uneven load.
At-a-glance visibility into the state of your clusters, and the strategy and progress of auto-balancing through a few key metrics

How Self-Balancing Simplifies Kafka Operations¶

Self-Balancing simplifies operational management of Kafka clusters in these ways:

When the load on the cluster is unevenly distributed, Self-Balancing automatically rebalances partitions to optimize performance
When a new broker is added to the cluster, Self-Balancing automatically fills it with partitions
When you want to remove a broker, you can use kafka-remove-brokers to shut down the broker and drain the partitions from it
When a broker has been down for a certain amount of time, Self-Balancing automatically reassigns the partitions to other brokers

Self-Balancing vs. Auto Data Balancer¶

Self-Balancing offers the following advantages over Auto Data Balancer

Cluster balance is continuously monitored so that rebalances run whenever they are needed
Broker failure conditions are detected and addressed automatically
No additional tools to run (built into the brokers)
Works with Confluent Control Center and offers a REST API
Much faster rebalancing (several orders of magnitude)

How it Works¶

Self-Balancing Clusters optimize Kafka load awareness. Resource usage within a cluster can be heterogeneous. Kafka out-of-the-box does not provide an automated process to improve cluster balance, but rather requires manual calculation of how to reassign partitions.

Self-Balancing simplifies this process by moving data to spread the cluster load evenly. Self-Balancing defines the meaning of “even” based on built-in goals, although you can provide optional input through configurations.

Architecture of a Self-Balancing Cluster¶

Kafka brokers collect metrics and and feed the data to an internal topic on the controller, which is on the lead broker for the cluster. The controller, therefore, plays a key role in cluster balancing. You can think of the controller as the node on which Self-Balancing is actively running.

The metrics are processed and decisions are made, based on goals.

This data is fed to other internal Kafka topics for monitoring and potential actions, such as generating a load balancing plan or triggering a rebalance.

Tip

You can view all topics, including internal topics, by using the kafka-topics --list command while Confluent Platform is running, for example: kafka-topics --list --bootstrap-server localhost:9092. Self-Balancing internal topics are prefixed with _confluent_balancer_.

Enabling Self-Balancing Clusters¶

For each broker in your cluster, set confluent.balancer.enable=true to enable Self-Balancing and make sure that this line is uncommented. Confluent Platform ships with Self-Balancing set to enabled in the example file: $CONFLUENT_HOME/etc/server.properties. To learn more, see confluent.balancer.enable.
To make Self-Balancing accessible for Configuration and Monitoring from Control Center, configure the Control Center cluster with REST endpoints to enable HTTP servers on the brokers, as described in Required Configurations for Control Center.

What defines a “balanced” cluster and what triggers a rebalance?¶

At a high level, Self-Balancing distinguishes between two types of imbalances:

Intentional operational actions, such as adding or removing brokers. In these cases, Self-Balancing saves the operator time and manual steps that would otherwise be required.
Ongoing cluster operations (such as a hot topic or partition). This balancing is ongoing throughout the life of the cluster.

These map to the two high level configuration options you have with Self-Balancing enabled:

Rebalance only when a broker is added
Rebalance anytime (for any uneven load, including changes in number of available brokers)

The first case is clear-cut. If brokers are added, self-healing occurs to redistribute data to the new broker or offload data from the missing broker.

The second case is more nuanced. To achieve ongoing cluster and data balance, Self-Balancing Clusters optimize on a number of goals and also avoid unnecessary movements if rebalancing would not materially improve cluster performance. Goals include considerations for replica placement and capacity, replication factors and throughput, multiple metrics on topics and partitions, leadership, rack awareness, disk usage and capacity, processor usage and capacity, network throughputs per broker, numerous load distribution targets, and more.

Self-Balancing Clusters employ continuous monitoring and data collection to track performance against these goals, and generate plans for rebalancing. A rebalance may or may not be triggered based on the implications of all weighted factors for the topology at a given moment.

Tip

In both cases, the cluster also rebalances when a broker is removed by user request (either from the command line or the Control Center), or if a broker goes missing some period of time (as specified by confluent.balancer.heal.broker.failure.threshold.ms).

Therefore, even with Self-Balancing set to Rebalance only when a broker is added, the cluster will rebalance for missing brokers unless you intentionally set properties to prevent this (for example, if confluent.balancer.heal.broker.failure.threshold.ms is set to -1).

What happens if the lead broker (controller) is removed or lost?¶

In a multi-broker cluster, one of the brokers is the leader or controller and plays a key role in Self-Balancing (as described in Architecture of a Self-Balancing Cluster). What happens if the broker where the controller is running is intentionally removed or crashes?

There is no impact to cluster integrity.
When the controller is removed or lost, a new controller is elected.
A broker removal request persists once it is made. If another broker becomes the controller, the new controller restarts and resumes the broker removal process, possibly with some delay.
The new leader will pick up the remove broker request and complete it.
If the lead broker is lost during an in-progress “add broker” operation, the “add broker” operation will not complete, and is marked as failed. If the new controller does not pick this up, you may need to restart the broker you were trying to add.

How do the brokers leverage Cruise Control?¶

Confluent Self-Balancing Clusters leverages Cruise Control for continuous metrics aggregation and reporting, reassignment algorithms and plans, and some rebalance triggers. Unlike Cruise Control, which has to be managed separately from Kafka, Self-Balancing is built into the brokers, meaning that data balancing is supported out of the box without additional dependencies. The result is custom-made, automated load balancing optimized for your Kafka clusters on Confluent Platform, and designed to work seamlessly with other components like Tiered Storage and Multi-Region Clusters.

What internal topics does Self-Balancing create and use?¶

See Self-Balancing Internal Topics in the Configuration and Commands Reference.

Limitations¶

Self-Balancing does not support JBOD (just a bunch of disks), also known as spanning, to make multiple disks appear as one.
If a broker contains the only replica of a partition, Self-Balancing will block elective attempts to remove the broker to prevent potential data loss (the broker removal operation will fail). The best way to remedy this is to increase replication factors on topics and internal topics on the broker you want to remove, as described in Configure replication factors for Self-Balancing in the tutorial.
Attempts to remove a broker immediately after cluster startup (while Self-Balancing is initializing) can fail due to insufficient metrics, and attempts to remove the lead broker can also fail at another phase. The solution is to retry broker removal after a period of time. If the broker is a controller, you must run broker removal from the command line, as it may not be available on the Control Center. To learn more, see Broker removal attempt fails during Self-Balancing initialization in Troubleshooting.

Configuration and Monitoring¶

Self-Balancing Clusters are self-managed, and can be enabled while the cluster is running. In most cases, you should not have to tinker with the defaults. Simply enable Self-Balancing either before or after you start the cluster, and allow it to auto-balance as needed.

In the example server.properties file that ships with Confluent Platform, confluent.balancer.enable is set to true, which means Self-Balancing is on.

Using Control Center¶

Tip

To make Self-Balancing accessible from Control Center, you must configure the Control Center cluster with REST endpoints to enable HTTP servers on the brokers, as described in Required Configurations for Control Center.
Information about working with Self-Balancing on Control Center is also the Self-Balancing Tutorial and the Control Center User Guide.
You cannot manage Self-Balancing through Control Center when it is deployed with Confluent Operator. To learn more about using Confluent Operator to manage Self-Balancing, see Scale Kafka clusters and balance data in the Operator documentation.

You can change these Self-Balancing settings from the Confluent Control Center (http://localhost:9021/) while the cluster is running:

Turn Self-Balancing on or off. In the example server.properties file that ships with Confluent Platform, confluent.balancer.enable is set to true, which means Self-Balancing is on. (See confluent.balancer.enable.)
Override the default throttle value (10485760 or 10MB/sec), which determines the maximum network bandwidth available for Self-Balancing. (confluent.balancer.throttle.bytes.per.second)
Toggle the trigger condition for rebalance to either Only when brokers are added (the default) or Anytime. (confluent.balancer.heal.uneven.load.trigger)

Select a cluster, click Cluster settings and select the Self-balancing tab.

The current settings for Self-Balancing are shown.
Click Edit Settings.
Make changes and click Save.

The updated settings are put into effect and reflected on the Self-balancing tab.

To view the property names for the Self-Balancing settings in effect (while editing or monitoring them), select Show raw configs.

Kafka Server Properties and Commands¶

In addition to the dynamic properties described above, more tuning parameters are exposed in $CONFLUENT_HOME/etc/kafka/server.properties. These are described in Configuration Options and Commands for Self-Balancing Clusters, specifically in the subtopic, Self-Balancing Configurations on the Brokers. To update the rest of these settings, you must stop the brokers and shut down the cluster.

See Self-Balancing Tutorial for an example of how to set up and run a quick test of Self-Balancing by removing a broker and monitoring the rebalance. The example provides guidance on proper configuration of replication factors and demos the Self-Balancing specific command, kafka-remove-brokers.

Metrics for Monitoring a Rebalance¶

Confluent Platform exposes several metrics through Java Management Extensions (JMX) that are useful for monitoring rebalances initiated by Self-Balancing:

The incoming and outgoing byte rate for reassignments is tracked by kafka.server:type=BrokerTopicMetrics,name=ReassignmentBytesInPerSec and kafka.server:type=BrokerTopicMetrics,name=ReassignmentBytesOutPerSec. These metrics are reported by each broker.
The number of pending and in-progress reassignment tasks currently tracked by Self-Balancing are tracked by kafka.databalancer:type=Executor,name=replica-action-pending and kafka.databalancer:type=Executor,name=replica-action-in-progress. These metrics are reported from the broker with the active data balancer instance (the controller).
The maximum follower lag on each broker is tracked by kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica. Reassigning partitions will cause this metric to jump to a value corresponding to the size of the partition and then slowly decay as the reassignment progresses. If this value slowly rises rather than slowly falling over time, the replication throttle is too low.

You can view the full list of metrics for Confluent Platform in Monitoring Kafka.

The Apache Kafka® documentation covers metrics reporting and monitoring by means of JMX endpoints here.

To monitor Self-Balancing, set the JMX_PORT environment variable before starting the cluster, then collect the reported metrics using your usual monitoring tools. JMXTrans, Graphite, and Grafana are a popular combination for collecting and reporting JMX metrics from Kafka. Datadog is another popular monitoring solution.

Troubleshooting¶

Following is a list of problems you may encounter while working with Self-Balancing and how to solve for them.

Self-Balancing options do not show up on Control Center¶

When Self-Balancing Clusters are enabled, status and configuration options are available on Control Center Cluster Settings > Self-balancing tab. If, instead, this tab displays a message about Confluent Platform version requirements and configuring HTTP servers on the brokers, this indicates something is missing from your configurations or that you are not running the required version of Confluent Platform.

Solution: Verify that you have the following settings and update your configuration as needed.

confluent.balancer.enable must be set to true to enable Self-Balancing.
In the Control Center properties file, confluent.controlcenter.streams.cprest.url must specify the associated URL for each broker in the cluster as REST endpoints for controlcenter.cluster, as described in Required Configurations for Control Center.
Your clusters must be deployed on Confluent Platform 6.0.0 or later.

Broker metrics are not displayed on Control Center¶

This issue is not specific to Self-Balancing, but related to proper configuration of multi-broker clusters, in general. You may encounter a scenario where Self-Balancing is enabled and displaying Self-Balancing options on Control Center, but broker metrics and per-broker drill-down and management options are not showing up on the Brokers Overview page. The most likely cause for this is that you did not configure the Metrics Reporter for Control Center. To do so, uncomment the following lines in properties files for all brokers. For example, in $CONFLUENT_HOME/etc/server.properties:

metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
confluent.metrics.reporter.bootstrap.servers=localhost:9092

Solution: To fix this on running clusters, you will need to shut down Control Center and the brokers, update the Metrics Reporter configurations for the brokers, and reboot.

This configuration is covered in more detail in the Self-Balancing tutorial at Enable the Metrics Reporter for Control Center.

Broker removal attempt fails during Self-Balancing initialization¶

Self-Balancing requires 15-20 minutes to initialize and collect metrics from brokers in the cluster. If you attempt to remove broker before metrics collection completes, the broker removal will fail almost immediately due to insufficient metrics for Self-Balancing. This is the most common use case for “remove broker” failing. The following error message will show on the command line or on Control Center, depending on which method you used for the remove operation:

Self-balancing requires a few minutes to collect metrics for rebalancing plans. Metrics collection is in process. Please try again after 900 seconds.

Solution: The solution for this is to wait about 15 minutes, and retry the broker removal.

If you want to remove a controller, the same factors are at play, so you should give some time for Self-Balancing to initialize before attempting a remove operation. In this case, though, the “remove broker” operation can fail at a later phase after the target broker is already shut down. In an offline state, the broker will no longer be accessible from Control Center. If this occurs, wait for 15 minutes, then retry the remove broker operation from the command line. Once Self-Balancing has initialized and had time to collect metrics, the operation should succeed, and the rebalancing plan will run.

To learn more about Self-Balancing initialization, see Self-Balancing Initialization.

Broker removal cannot complete due to offline partitions¶

Broker removal can also fail in cases where taking a broker down will result in having fewer online brokers than the number of replicas required in your configurations.

The broker status (available with kafka-remove-brokers --describe) will remain as follows, until you restart one or more of the offline brokers:

[2020-09-17 23:40:53,743] WARN [AdminClient clientId=adminclient-1] Connection to node -5 (localhost/127.0.0.1:9096) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)Broker 1 removal status:
Partition Reassignment: IN_PROGRESS
Broker Shutdown: COMPLETE

A short-hand way of troubleshooting this is to ask “how many brokers are down?” and “how many replicas/replication factors must the cluster support?”

Partition reassignment (the last phase in broker removal) will fail to complete in any case where you have n brokers down, and your configuration requires n + 1 or more replicas.

Alternatively, you can consider how many online brokers you need to support the required number of replicas. If you have n brokers online, these can support at most a total of n replicas.

Solution: The solution is to restart the down brokers, and perhaps modify the cluster configuration as a whole. This might include both adding brokers and modifying replicas/replication factors (see example below).

Scenarios that lead to this problem can be a combination of under-replicated topics and topics with too many replicas for the number of online brokers. Having a topic with a replication factor of 1 does not necessarily lead to a problem in and of itself.

A quick way to get an overview of configured replicas on a running cluster is to use kafka-topics --describe on a specified topic, or on the whole cluster (with no topic specified). For system topics, you can scan the replication factors and replicas on system properties (which generate system topics). The Self-Balancing Tutorial covers these commands, replicas/replication factors, and the impact of these configurations.

Too many excluded topics causes problems with Self-Balancing¶

Excluding too many topics (and by inference, partitions) can be counter-productive to maintaining a well-balanced cluster. In internal tests, excluding system topics which accounted for approximately 100 out of 500 total topics was enough to put Self-Balancing into a less than optimal state.

The manifest result of this is that the cluster constantly churns on rebalancing.

Solution: Reduce the number of excluded topics. The relevant configuration options to modify these settings are:ref:sbc-config-exclude-topic-names and confluent.balancer.exclude.topic.prefixes.