Kafka Cruise Control 3

Overview

Kafka Cruise Control 3 helps you run Kafka clusters at a large scale. It helps automate Kafka operations, such as monitoring cluster workload, rebalancing a cluster based on predefined constraints, and detecting and fixing anomalies. It comprises four components (Load Monitor, Analyser, Anomaly Detector, and Executor) and a REST API.

Architecture

The architecture of cruise-control3 is illustrated below.

1. REST API: Cruise-control3 offers a REST API enabling users to query Kafka cluster load, optimize proposals, and trigger administrative tasks.

2. Load Monitor: Collects Kafka metrics and derives per-partition resource metrics, forming a cluster load model crucial for optimization.

3. Load Monitor Workflow:

   • Metric Fetcher Manager: Coordinates metric sampling tasks with configurable threads.
   • Sampling Tasks: Includes metric sampling, bootstrap, and linear model training executed by metric fetcher threads.
   • Metric Sampler Partition Assignor: Ensures even partition distribution among fetchers, considering topic and load balancing.

4. Cluster Load Model: Reflects the current cluster state, providing interfaces to simulate partition movements for optimization.

5. Sample Store: Saves metric and training samples to external storage for future reference, facilitating historical analysis.

6. Analyzer: Utilizes heuristic methods to generate optimization proposals based on user-defined goals and the cluster load model.

7. Anomaly Detector: Identifies and addresses anomalies such as broker failures, goal violations, disk failures, metric anomalies, and topic anomalies.

8. Executor: Executes optimization proposals, ensuring resource-awareness.

Quickstart

1. Cruise-control3 comes as a part of Kafka3 components, It can be installed on one of the nodes within the Kafka3 cluster through Ambari.
2. Using the ADD Component, you can add cruise-control3 on one of the nodes as a component either at the time of Kafka installation or after the Kafka-broker installation. Note: If you are installing cruise-control3 after the Kafka-broker installation, for Kafka to pick up the configurations related to cruise-control3, you need to restart Kafka brokers before starting cruise-control3.
3. Select Kafka as a service under the Choose Services section.

4. Assign the nodes on which you want to install Kafka broker:

5. Select the node on which you want to add cruise-control3 as a component:

6. Make the following configuration changes related to cruise-control3:
   1. Under Advanced cruise-control3-env and Advanced kafka-env, mention the java implementation to use:

b. By default cruise-control3 runs on port 9096, If you want to run cruise-control3 on some other port, need to make the following changes:

• Under Advanced cruise-control3, change webserver.http.port to the desired port, also change the port number in Advanced cruise-control3-env and Advanced cruise-control3-ui-config through ambari.

c. Under Advanced cruise-control3-ui-config, edit the localhost to the hostname on which cruise-control3 is installed. This configuration is related to the cruise-control3 UI. If SSL is enabled, change http to https in Advanced cruise-control3-ui-config.

• Format of config.csv: config.csv is a CSV file with three columns separated by commas:
  • First column_:_ Logical Group Name (Examples: dev, production, trade)
  • Second column_:_ cruise-control3 Instance Name (Examples: finance, logging, )
  • Third column: URL path to the cruise-control3 instances (Examples: /kafkacruisecontrol/, http://dev.acceldata.ce:9096/kafkacruisecontrol)

Advanced cruise-control3-ui-configuration

7. This step is only applicable if kafka3 is running in kraft mode:
   • If you are using cruise-control in kraft mode, make the following changes:
     • Add the following property under Advanced cruise-control3.

Advanced cruise-control3

• Remove zookeeper.connect and set zookeeper.security.enabled as false under Advanced cruise-control3 , also remove any configurations related to zookeeper.

8. Click on DEPLOY to deploy the Kafka brokers and cruise-control3 on the selected nodes.

9. Once the deployment is completed, Cruise-control3 and Kafka broker can be installed on the specified nodes:

Security

SSL Enablement

Update the following properties as per your SSL configurations.

Advanced cruise-control3

Advanced kafka3-broker

Advanced cruise-control3-ui-configuration

Kerberos Configuration

1. Cruise-control3 uses Kafka user as its default user. So, as such no specific kerberos configuration changes are required for cruise-control3.
2. The Advanced cruise-control3-jaas-conf file can also be managed by Ambari.

Configuring JWT authentication with Apache Knox

1. Create a credentials file containing the following information (for instance in /tmp/test-roles.credentials:

kafka: ,ADMIN

sam: ,VIEWER

2. Make the following changes in Advanced cruise-control through Ambari:

jwt.auth.certificate.location=/path/to/gateway-identity.pem

jwt.authentication.provider.url=https://cent1.acceldata.ce:8443/gateway/knoxsso/api/v1/websso?originalUrl=http://cent1.acceldata.ce:9095

jwt.cookie.name=hadoop-jwt

webserver.auth.credentials.file=/path/to/test-roles.credentials

webserver.security.enable=true

webserver.security.provider=com.linkedin.kafka.cruisecontrol.servlet.security.jwt.JwtSecurityProvider

3. In the Advanced knoxsso-topology template, you can edit the value of knoxsso.token.ttl.By default, this value is set to 30000ms, meaning the Time To Live (TTL) for tokens used in Knox Single Sign-On (SSO) authentication expires after 30 seconds. You can increase this value to your desired duration.

4. Open the URL mentioned in Advanced cruise-control-ui-config, it must redirect you to the Knox auth page (then log in with the admin/admin-password or username/password pair). It must redirect to the CC page.

Pluggable Components

Metric Sampler

The metric sampler is one of the most important pluggables in Kafka Cruise Control, it allows user to easily deploy Cruise Control to various environment and work with the existing metric system.

The default implementation of metric sampler is reading the broker metrics produced by CruiseControlMetricsReporter on the broker. This is assuming that users are running Kafka brokers by setting the metric.reporters configuration on the Kafka brokers to be com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter.

Metric Sampler Partition Assignor

When users have multiple metric sampler threads, the metric sampler partition assignor is responsible for assigning partitions to the metric samplers. This is useful when users have some existing metric system. The default implementation assigns all the partitions of the same topic to the same metric sampler.

Sample Store

The Sample Store is used to store the collected metric samples and training samples to external storage. One problem in metric sampling is that we are using some derived data from the raw metrics. And the way we derive the data relies on the metadata of the cluster at that moment. So when we look at the old metrics, if we do not know the metadata at the point the metric was collected the derived data would be inaccurate. Sample Store helps solve this problem by storing the derived data directly to an external storage for later loading.

The default implementation of Sample Store produces the samples back to the Kafka topic.

Broker Capacity Config Resolver

The broker Capacity Config Resolver is the way for Cruise Control to get the broker capacity for each of the resources. The default implementation is file based properties. Users can also have a customized implementation to retrieve the capacity of the brokers from some hardware resource management system.

Goals

The goals in Kafka Cruise Control are pluggable with different priorities.

• Rack-awareness: A goal that ensures all the replicas of each partition are assigned in a rack aware manner.

• RackAwareDistributionGoal - Contrary to RackAwareGoal, as long as replicas of each partition can achieve a perfectly even distribution across the racks, this goal lets placement of multiple replicas of a partition into a single rack.

• ReplicaCapacityGoal: Attempts to make all the brokers in a cluster to have less than a given number of replicas.

• CapacityGoals: Goals that ensures the broker resource utilization is below a given threshold for the corresponding resource. Capacity goals are:

  • DiskCapacityGoal
  • NetworkInboundCapacityGoal
  • NetworkOutboundCapacityGoal
  • CpuCapacityGoal

• ReplicaDistributionGoal: Attempts to make all the brokers in a cluster to have the similar number of replicas.

• PotentialNwOutGoal: A goal that ensures the potential network output (when all the replicas becomes leaders) on each of the broker do not exceed the broker’s network outbound bandwidth capacity.

• ResourceDistributionGoals: Attempts to make the resource utilization variance among all the brokers are within a certain range. This goal does not do anything if the cluster is in a low utilization mode (when all the resource utilization of each broker is below a configured percentage). This is not a single goal, but consists of the following separate goals for each of the resources.

  • DiskUtilDistributionGoal
  • NetworkInboundUtilDistributionGoal
  • NetworkOutboundUtilDistributionGoal
  • CpuUtilDistributionGoal

• TopicReplicaDistributionGoal: Attempts to make the replicas of the same topic evenly distributed across the entire cluster.

• LeaderReplicaDistributionGoal: Attempts to make all the brokers in a cluster to have the similar number of leader replicas.

• LeaderBytesInDistributionGoal: Attempts to make the leader bytes in rate on each host to be balanced.

• PreferredLeaderElectionGoal: Attempts to make the first replica in replica list leader replica of the partition for all topic partition.

• KafkaAssignerGoals: These goals can picked up if kafka_assigner parameter is set to true in the corresponding request.

  • KafkaAssignerDiskUsageDistributionGoal: A goal that ensures all the replicas of each partition are assigned in a rack aware manner.
  • KafkaAssignerEvenRackAwareGoal: Attempts to make all the brokers in a cluster to have the similar number of replicas.

• IntraBrokerDiskCapacityGoal: Goals that ensures the disk resource utilization is below a given threshold. This goal will be pick up if rebalance_disk parameter is set to true in rebalance request.

• IntraBrokerDiskUsageDistributionGoal: Attempts to make the utilization variance among all the disks within same broker are within a certain range. This goal can be picked up if rebalance_disk parameter is set to true in rebalance request.

• BrokerSetAwareGoal: BrokerSet is defined as a subset of brokers in the cluster. This Goal attempts to make the replica movements constrained within the boundary of a BrokerSet.

Cruise Control REST API Endpoints

You can manage and monitor Cruise Control through its REST API, which offers GET and POST endpoints. Knowing the specific action or query you want to perform is essential when crafting a curl command to interact with the Cruise Control REST API.

Cruise Control's REST API includes two main types of endpoints:

• GET: The GET endpoints allow you to retrieve details concerning the rebalancing operations, the current state of Cruise Control, and the status of Kafka brokers. These endpoints perform read-only functions, meaning they do not affect Cruise Control, Kafka, or the ongoing rebalancing activities.
• POST: With POST endpoints, you have the capability to alter the rebalancing operations, adjust the count of Kafka brokers, set configurations for Cruise Control, and tailor specific tasks related to sampling and proposals.

By utilizing these endpoints appropriately, you can effectively control and check the status of Cruise Control using simple command line tools.

Using the Cruise Control REST API endpoints

The information provided below offers a brief overview of the operations available through the GET and POST endpoints of the Cruise Control REST API. This summary is intended to give a quick understanding of the endpoints functions. For the detailed description of the Cruise Control REST API, see the official REST API documentation.

GET Requests

The GET requests in Kafka Cruise Control REST API are for read only operations, i.e. the operations that do not have any external impacts. The GET requests include the following operations:

• Query the state of Cruise Control
• Query the current cluster load
• Query partition resource utilization
• Query partition and replica state
• Get optimization proposals
• Query the user request result

Query the state of Cruise Control

User can query the state of Kafka Cruise Control at any time by issuing a HTTP GET request.

GET /kafkacruisecontrol/state

Example:

GET 'http://hostname:9095/kafkacruisecontrol/state'

Description: You can check the state of Kafka Cruise Control, which encompasses the status of the monitor, executor, analyzer, and anomaly detector components.

NOTE: If verbose is set to true, the details about monitored windows and goals are displayed.

Query the current cluster load

Once Cruise Control Load Monitor shows it is in the RUNNING state, you can use the following HTTP GET to get the cluster load:

GET /kafkacruisecontrol/load

Example:

GET 'http://hostname:9095/kafkacruisecontrol/load'

Description: You can query the current cluster load using Kafka Cruise Control. For accurate results, the Load Monitor must be in the RUNNING state. This query provides information on the load per broker and load per host. It only displays data for valid partitions that have a sufficient number of metric samples.

Query partition resource utilization

The following GET request gives the partition load sorted by the utilization of a given resource:

GET /kafkacruisecontrol/partition_load

Example:

GET 'http://hostname:9095/kafkacruisecontrol/partition_load'

Description: You can query the resource utilization of partitions in Kafka Cruise Control. The partition load data is organized and presented based on the specified resource type.

The returned result would be a partition list sorted by the utilization of the specified resource in the time range specified by start and end. The resource can be CPU, NW_IN, NW_OUT and DISK. By default, the start is the earliest monitored time, the end is current wall clock time, resource is DISK, and entries is the all partitions in the cluster.

Query partition and replica state

The following GET request gives partition healthiness on the cluster:

GET /kafkacruisecontrol/kafka_cluster_state

Example:

GET 'http://hostname:9095/kafkacruisecontrol/kafka_cluster_state'

Description: You can query the state of partitions and replicas in Kafka Cruise Control. This query provides detailed information for each broker and each partition, helping you understand their current operational status.

The returned result contains the following information

• For each broker

  • Distribution of leader/follower/out-of-sync/offline replica information
  • Online/offline disks

• For each partition

  • Distribution of leader/follower/in-sync/out-of-sync/offline replica information

Get optimization proposals

The following GET request returns the optimization proposals generated based on the workload model of the given timestamp. The workload summary before and after the optimization can also be returned.

GET /kafkacruisecontrol/proposals

Example:

GET 'http://hostname:9095/kafkacruisecontrol/proposals'

Description: You can retrieve the optimization proposals generated based on the workload model in Kafka Cruise Control. These proposals can be generated either using the snapshot window or focusing on valid partitions, providing insights into potential optimization strategies.

A proposal can be generated based on valid_window or valid_partitions.

• valid_windows:

  • Rebalance the cluster based on the information in the available valid snapshot windows. A valid snapshot window is a window whose valid monitored partitions coverage meets the requirements of all the goals. (This is the default behavior).

• valid_partitions:

  • Rebalance the cluster based on all the available valid partitions. All the snapshot windows can be included in this case.

Users can only specify either valid_windows or valid_partitions, but not both.

If verbose is turned on, Cruise Control returns all the generated proposals. Otherwise a summary of the proposals is returned.

Users can specify excluded_topics to prevent certain topics replicas from moving in the generated proposals.

If use_ready_default_goals is turned on, Cruise Control uses whatever ready goals (based on available metric data) to calculate the proposals.

Query the user request result

The following get request allows the user to get a full list of all the active or completed (and not recycled) tasks inside Cruise Control, with their initial request detail (request time/IP address/request URL and parameter) and UUID information. User can then use the returned UUID and URL to fetch the original final result of the specific request.

GET /kafkacruisecontrol/user_tasks

Example:

GET 'http://hostname:9095/kafkacruisecontrol/user_tasks'

Description: You can retrieve the results of user requests in Kafka Cruise Control. This includes a comprehensive list of all active and completed tasks, offering visibility into the ongoing and historical activities within Cruise Control.

User can use user_task_ids/client_ids/endpoints/types and make Cruise Control only return requests they are interested. By default all the requests get returned.

If fetch_completed_task is set to true, the original response of each request is returned. In the case where a task is completed with errors, the response can be CompletedWithError.

POST Requests

The post requests of Kafka Cruise Control REST API are operations that can have impact on the Kafka cluster. The post operations include:

• Trigger a workload balance
• Add a list of new brokers to Kafka Cluster
• Decommission a list of brokers from the Kafka cluster
• Fix offline replicas in Kafka cluster
• Demote a list of brokers from the Kafka cluster
• Stop the current proposal execution task
• Pause metrics load sampling
• Resume metrics load sampling
• Change Kafka topic configuration
• Change Cruise Control configuration

Trigger a workload balance

The following POST request lets Kafka Cruise Control rebalance a Kafka cluster.

POST /kafkacruisecontrol/rebalance

Description: You have the capability to initiate a workload balance for a Kafka cluster through Kafka Cruise Control. This can be based on either valid windows or valid partitions. Additionally, you can specify the optimization goal you wish to pursue. By default, the dry-run mode is activated, providing a simulation of the rebalancing process without actually executing it. To trigger the rebalancing process, you need to set dryrun=false.

Add a list of new brokers to Kafka Cluster

The following POST request adds the given brokers to the Kafka cluster.

POST /kafkacruisecontrol/add_broker?brokerid=[id1,id2...]

Description: You can add a list of brokers to a Kafka cluster using Kafka Cruise Control. This process involves relocating replicas from existing brokers to the newly added brokers. Optionally, you can throttle the movement of replicas to the newly added brokers, with the throttle being applied on the existing brokers. By default, the dry-run mode is activated, providing a simulation of the process without actual execution. To initiate the addition of brokers, you need to set dryrun=false.

When adding new brokers to a Kafka cluster, Cruise Control makes sure that the replicas will only be moved from the existing brokers to the provided new broker, but not moved among existing brokers.

Decommission a list of brokers from the Kafka cluster

The following POST request removes a list of brokers from the Kafka cluster:

POST /kafkacruisecontrol/remove_broker?brokerid=[id1,id2...]

Description: You can remove a list of brokers from a Kafka cluster using Kafka Cruise Control. This process involves transferring partitions from the brokers slated for removal to other existing brokers. You have the option to specify the destination broker for these partitions. Additionally, you can throttle the removed brokers during the partition movement. By default, the dry-run mode is activated, providing a simulation of the process without actual execution. To initiate the removal of brokers, you need to set dryrun=false.

Fix offline replicas in Kafka cluster

The following POST request moves all the offline replicas from dead disks or brokers.

POST /kafkacruisecontrol/fix_offline_replicas

Description: You can resolve offline replicas in a Kafka cluster using Kafka Cruise Control. If the specified topic contains offline replicas, Cruise Control will move them to healthy brokers. By default, the process runs in dry-run mode, simulating the actions without actually executing them. To actively fix offline replicas, you need to set dryrun=false.

Demote a list of brokers from the Kafka cluster

The following POST request moves all the leader replicas away from a list of brokers.

POST /kafkacruisecontrol/demote_broker?brokerid=[id1, id2...]

Description: You can transfer all leader replicas from a specified list of brokers using Kafka Cruise Control. This action demotes all replicas on the designated brokers to be preferred the least, prompting a preferred leader replica election to migrate the leader replicas away from those brokers. By default, this operation runs in dry-run mode, providing a simulation of the process without actually executing it. To actively demote brokers and trigger the leader replica migration, you need to set dryrun=false.

User can also request to move all the leader replicas away from the a list of disks.

POST /kafkacruisecontrol/demote_broker?brokerid_and_logdirs=[id1-logdir1, id2-logdir2...]

Description: You have the capability to relocate all leader replicas from a specified list of disks using Kafka Cruise Control. This action entails designating all replicas on the specified disks to be least preferred, subsequently initiating a preferred leader replica election to migrate the leader replicas away from those disks.

Demoting a broker or disk is consist of tow steps.

• Make all the replicas on given broker or disk the least preferred replicas for leadership election within their corresponding partitions.
• Trigger a preferred leader election on the partitions to migrate the leader replicas off the broker or disk.

Stop the current proposal execution task

The following POST request will let Kafka Cruise Control stop an ongoing rebalance, add_broker, remove_broker, fix_offline_replica, topic_configuration or demote_broker operation:

POST /kafkacruisecontrol/stop_proposal_execution

Description: You can halt an ongoing proposal execution task using Kafka Cruise Control. This allows you to stop a proposal while the task is still in progress.

Pause metrics load sampling

The following POST request lets Kafka Cruise Control pause an ongoing metrics sampling process:

POST /kafkacruisecontrol/pause_sampling

Description: You can pause an ongoing metrics sampling process in Kafka Cruise Control. The reason for the pause is recorded and can be viewed when querying the state of Cruise Control.

The reason to pause sampling will be recorded and shows up in state endpoint(under LoadMonitor sub state).

Resume metrics load sampling

The following POST request lets Kafka Cruise Control resume a paused metrics sampling process:

POST /kafkacruisecontrol/resume_sampling

Description: You can resume a paused metrics load sampling process in Kafka Cruise Control. When resuming, the reason for the resumption is recorded and can be viewed when querying the state of Cruise Control.

The reason to resume sampling can be recorded and shows up in state endpoint(under LoadMonitor sub state).

Change Kafka topic configuration

Currently Cruise Control only supports changing topic's replication factor via topic_configuration endpoint. Ultimately we want make Cruise Control the central place to change any topic configurations(partition count, retention time etc.).

The following POST request can change the topic's replication factor.

POST /kafkacruisecontrol/topic_configuration?topic=[topic_regex]&replication_factor=[target_replication_factor]

Changing topic's replication factor do not move any existing replicas. goals are used to determine which replica to be deleted (to decrease topic's replication factor) and which broker to assign new replica (to increase topic's replication factor).

Change Cruise Control configuration

Some Cruise Control configs can be changed dynamically via admin endpoint, which includes

• Dynamically change the partition and leadership concurrency and the interval between checking and updating (if needed) the progress of an ongoing execution.
• Enable or disable self-healing for the specified anomaly types.
• Drop selected recently removed or demoted brokers.
• Enable or disable the specified concurrency adjusters.
• Enable or disable the (At/Under) MinISR-based concurrency adjustment.

To Enable or disable self-healing, send the POST request as shown below:

POST /kafkacruisecontrol/admin?disable_self_healing_for=[anomaly_type]

To increase or decrease the execution concurrency, send the POST request as shown below:

POST /kafkacruisecontrol/admin?concurrent_partition_movements_per_broker=[integer]

To drop recently removed or demoted brokers, send the POST request as shown below:

POSTs /kafkacruisecontrol/admin?drop_recently_removed_brokers=[broker_ids]

Two Step Verification for POST Requests

The POST requests enable users to perform various admin operations for cluster maintenance in Kafka clusters. However, if users mistype a command, it is possible to start an unintended execution (e.g. remove broker-X rather than broker-Y).

Two-step verification aims to help users verify the command they (or their peers) intend to run by letting them review requests explicitly to approve or discard them, and enable execution of only the approved requests.

How can I enable two-step verification for POST requests?

Set two.step.verification.enabled configuration to true in Advanced cruise-control through Ambari.

What are the POST endpoints, whose requests require a two-step verification?

1. add_broker,
2. remove_broker,
3. fix_offline_replicas
4. rebalance,
5. stop_proposal_execution
6. pause_sampling
7. resume_sampling
8. demote_broker
9. admin

NOTE : 2-step verification aims to help users verify the command they (or their peers) intend to run by letting them review requests explicitly to approve or discard them, and enable execution of only the approved requests. Read 2 step verification for POST requests for more detail.

Cruise Control Frontend (CCFE)

Frontend for Kafka Cruise Control offers a centralized dashboard with various features. These are some of the features of the application:

1. Kafka Cluster Status
2. Kafka Cluster Load
3. Kafka Partition Load
4. Cruise Control State
5. Cruise Control Tasks
6. Kafka Cluster Administration
7. Cruise Control Proposals
8. Cruise Control Tasks
9. Peer review

In addition to these, there is support for these common features:

1. Running actions in Dry Run (Safe Mode) by default.
2. All Cruise Control REST API parameters are exposed as Input controls.
3. Safe Default values are provided to execute the actions as fast as possible.
4. Understand the Async Response from Cruise Control and render the progress.
5. Display URLs of the End points for every action.
6. Display Full Stack Trace responses from Cruise Control instead of shorter ones.

You can access the Cruise control UI using the Quick Links section of Kafka.

Refresh the following pages using the Refresh button. Upon refreshing, the browser might throw an error as shown on the image.

• Kafka Cluster Load
• Cruise Control Proposals
• Peer review

If you encounter the above issue, navigate to the Kafka Cluster State page and then return to the page where you have encountered this issue. This action must resolve it.

Application Screenshots

Kafka Cluster Status

This page shows the real-time status of the Kafka cluster and reports any URP, Offline Partitions, Disks, etc.

Kafka Cluster Load

This page shows the load on all the brokers as calculated by Cruise Control.

Cruise Control Status

This page shows the internal status of the Cruise Control (like Monitor, Executor, Analyzer, and Anomaly Detector).

User Tasks

This page shows all the past or current tasks that Cruise Control got.

Administration

This page show all the brokers in the system along with the possible administrative actions (Add Broker, Remove Broker, Demote Broker, and Rebalance, etc.).

Below screen is for Doing Kafka Cluster rebalance with advanced options.

Reference

• Cruise Control
• Cruise Control UI