Using Streams for Apache Kafka on RHEL in KRaft mode

Procedure

1. Install RHEL in FIPS mode.

   For further information, see the information on security hardening in the RHEL documentation.

2. Proceed with the installation of Streams for Apache Kafka.

3. Configure Streams for Apache Kafka to use FIPS-compliant algorithms and protocols.

   If used, ensure that the following configuration is compliant:

   • SSL cipher suites and TLS versions must be supported by the JDK framework.
   • SCRAM-SHA-512 passwords must be at least 32 characters long.

1. Add a new kafka user and group:

   groupadd kafka
   useradd -g kafka kafka
   passwd kafka

2. Extract and move the contents of the amq-streams-<version>-kafka-bin.zip file into the /opt/kafka directory:

   unzip amq-streams-<version>-kafka-bin.zip -d /opt
   mv /opt/kafka*redhat* /opt/kafka

3. Change the ownership of the /opt/kafka directory to the kafka user:

   chown -R kafka:kafka /opt/kafka

4. Create directory /var/lib/kafka for storing Kafka data and set its ownership to the kafka user:

   mkdir /var/lib/kafka
   chown -R kafka:kafka /var/lib/kafka

   You can now run a default configuration of Kafka as a single-node cluster.

   You can also use the installation to run other Kafka components, like Kafka Connect, on the same host.

   To run other components, specify the hostname and port to connect to the Kafka broker using the bootstrap.servers property in the component configuration.

   Example bootstrap servers configuration pointing to a single Kafka broker on the same host

   bootstrap.servers=localhost:9092

   However, we recommend installing and running Kafka components on separate hosts.

5. (Optional) Install Kafka components on separate hosts.

   1. Extract the installation files to the /opt/kafka directory on each host.
   2. Change the ownership of the /opt/kafka directory to the kafka user.

   3. Add bootstrap.servers configuration that connects the component to the host (or hosts in a multi-node environment) running the Kafka brokers.

      Example bootstrap servers configuration pointing to Kafka brokers on different hosts

      bootstrap.servers=kafka0.<host_ip_address>:9092,kafka1.<host_ip_address>:9092,kafka2.<host_ip_address>:9092

      You can use this configuration for Kafka Connect, MirrorMaker 2, and the Kafka Bridge.

Procedure

1. Generate a unique ID for the Kafka cluster.

   You can use the kafka-storage tool to do this:

   /opt/kafka/bin/kafka-storage.sh random-uuid

   The command returns an ID. A cluster ID is required in KRaft mode.

2. Create a configuration properties file for each node in the cluster.

   You can base the file on the examples provided with Kafka.

   1. Specify a role as broker, controller, or broker, controller

      For example, specify process.roles=broker, controller for a combined role.

   2. Specify a unique node.id for each node in the cluster starting from 0.

      For example, node.id=1.

   3. Specify a list of controller.quorum.voters in the format <node_id>@<hostname:port>.

      For example, controller.quorum.voters=1@localhost:9093.

   4. Specify listeners:

      • Configure the name, hostname and port for each listener.

        For example, listeners=PLAINTEXT:localhost:9092,CONTROLLER:localhost:9093.

      • Configure the listener names used for inter-broker communication.

        For example, inter.broker.listener.name=PLAINTEXT.

      • Configure the listener names used by the controller quorum.

        For example, controller.listener.names=CONTROLLER.

      • Configure the name, hostname and port for each listener that is advertised to clients for connection to Kafka.

        For example, advertised.listeners=PLAINTEXT:localhost:9092.

3. Set up log directories for each node in your Kafka cluster:

   /opt/kafka/bin/kafka-storage.sh format -t <uuid> -c /opt/kafka/config/kraft/server.properties

   Returns:

   Formatting /tmp/kraft-combined-logs

   Replace <uuid> with the cluster ID you generated. Use the same ID for each node in your cluster.

   Apply the broker configuration using the properties file you created for the broker.

   By default, the log directory (log.dirs) specified in the server.properties configuration file is set to /tmp/kraft-combined-logs. The /tmp directory is typically cleared on each system reboot, making it suitable for development environments only.

   You can add a comma-separated list to set up multiple log directories.

4. Start each Kafka node.

   /opt/kafka/bin/kafka-server-start.sh /opt/kafka/config/kraft/server.properties

5. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   process ID kafka.Kafka /opt/kafka/config/kraft/server.properties

   Check the logs of each node to ensure that they have successfully joined the KRaft cluster:

   tail -f /opt/kafka/logs/server.log

Procedure

1. Stop the Kafka node.

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

2. Confirm that the Kafka node is stopped.

   jcmd | grep kafka

1. Stop the Kafka broker:

   /opt/kafka/bin/kafka-server-stop.sh

2. Make any changes to the broker configuration that require a restart after completion.

   For further information, see the following:

   • Configuring Kafka
   • Upgrading Kafka nodes

3. Restart the Kafka broker:

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

4. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   process ID kafka.Kafka /opt/kafka/config/kraft/server.properties

   Check the logs of each node to ensure that they have successfully joined the KRaft cluster:

   tail -f /opt/kafka/logs/server.log

5. Wait until the broker has zero under-replicated partitions. You can check from the command line or use metrics.

   • Use the kafka-topics.sh command with the --under-replicated-partitions parameter:

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port>  --describe --under-replicated-partitions

     For example:

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions

     The command provides a list of topics with under-replicated partitions in a cluster.

     Topics with under-replicated partitions

     Topic: topic3 Partition: 4 Leader: 2 Replicas: 2,3 Isr: 2
     Topic: topic3 Partition: 5 Leader: 3 Replicas: 1,2 Isr: 1
     Topic: topic1 Partition: 1 Leader: 3 Replicas: 1,3 Isr: 3
     # …

     Under-replicated partitions are listed if the ISR (in-sync replica) count is less than the number of replicas. If a list is not returned, there are no under-replicated partitions.

   • Use the UnderReplicatedPartitions metric:

     kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

     The metric provides a count of partitions where replicas have not caught up. You wait until the count is zero.

     Tip

     Use the Kafka Exporter to create an alert when there are one or more under-replicated partitions for a topic.

Procedure

1. Add the Environment Variables Configuration Provider JAR file to the Kafka libs directory.

2. Initialize the Environment Variables Configuration Provider in the configuration properties file of the Kafka component. For example, to initialize the provider for Kafka, add the configuration to the server.properties file.

   Configuration to enable the Environment Variables Configuration Provider

   config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider

3. Add configuration to the properties file to load data from environment variables.

   Configuration to load data from an environment variable

   option=${env:<MY_ENV_VAR_NAME>}

   Use capitalized or upper-case environment variable naming conventions, such as MY_ENV_VAR_NAME.

4. Save the changes.

5. Restart the Kafka component.

   For information on restarting brokers in a multi-node cluster, see Section 3.6, “Performing a graceful rolling restart of Kafka brokers”.

Procedure

1. Write the OPA policies required for authorizing client requests to perform operations on the Kafka brokers.

   See Defining OPA policies.

   Now configure the Kafka brokers to use OPA.

2. Install the OPA authorizer plugin for Kafka.

   See Connecting to the OPA.

   Make sure that the plugin files are included in the Kafka classpath.

3. Add the following to the Kafka server.properties configuration file to enable the OPA plugin:

   authorizer.class.name: com.bisnode.kafka.authorization.OpaAuthorizer

4. Add further configuration to server.properties for the Kafka brokers to access the OPA policy engine and policies.

   For example:

   opa.authorizer.url=https://OPA-ADDRESS/allow 1
   opa.authorizer.allow.on.error=false 2
   opa.authorizer.cache.initial.capacity=50000 3
   opa.authorizer.cache.maximum.size=50000 4
   opa.authorizer.cache.expire.after.seconds=600000 5
   super.users=User:alice;User:bob 6

   1

   (Required) The OAuth 2.0 token endpoint URL for the policy the authorizer plugin will query. In this example, the policy is called allow.

   2

   Flag to specify whether a client is allowed or denied access by default if the authorizer plugin fails to connect with the OPA policy engine.

   3

   Initial capacity in bytes of the local cache. The cache is used so that the plugin does not have to query the OPA policy engine for every request.

   4

   Maximum capacity in bytes of the local cache.

   5

   Time in milliseconds that the local cache is refreshed by reloading from the OPA policy engine.

   6

   A list of user principals treated as super users, so that they are always allowed without querying the Open Policy Agent policy.

   Refer to the Open Policy Agent website for information on authentication and authorization options.

5. Verify the configured permissions by accessing Kafka brokers using clients that have and do not have the correct authorization.

Creating a topic

1. Create a topic using the kafka-topics.sh utility and specify the following:

   • Host and port of the Kafka broker in the --bootstrap-server option.
   • The new topic to be created in the --create option.
   • Topic name in the --topic option.
   • The number of partitions in the --partitions option.

   • Topic replication factor in the --replication-factor option.

     You can also override some of the default topic configuration options using the option --config. This option can be used multiple times to override different options.

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --create --topic <TopicName> --partitions <NumberOfPartitions> --replication-factor <ReplicationFactor> --config <Option1>=<Value1> --config <Option2>=<Value2>

     Example of the command to create a topic named mytopic

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic mytopic --partitions 50 --replication-factor 3 --config cleanup.policy=compact --config min.insync.replicas=2

2. Verify that the topic exists using kafka-topics.sh.

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --describe --topic <TopicName>

   Example of the command to describe a topic named mytopic

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

Describing a topic

1. Describe a topic using the kafka-topics.sh utility and specify the following:

   • Host and port of the Kafka broker in the --bootstrap-server option.
   • Use the --describe option to specify that you want to describe a topic.
   • Topic name must be specified in the --topic option.

   • When the --topic option is omitted, it describes all available topics.

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --describe --topic <topic_name>

     Example of the command to describe a topic named mytopic

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic mytopic

     The command lists all partitions and replicas which belong to this topic. It also lists all topic configuration options.

Modify topic configuration

1. Use the kafka-configs.sh tool to get the current configuration.

   • Specify the host and port of the Kafka broker in the --bootstrap-server option.
   • Set the --entity-type as topic and --entity-name to the name of your topic.

   • Use --describe option to get the current configuration.

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --describe

     Example of the command to get configuration of a topic named mytopic

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --describe

2. Use the kafka-configs.sh tool to change the configuration.

   • Specify the host and port of the Kafka broker in the --bootstrap-server option.
   • Set the --entity-type as topic and --entity-name to the name of your topic.
   • Use --alter option to modify the current configuration.

   • Specify the options you want to add or change in the option --add-config.

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --alter --add-config <option>=<value>

     Example of the command to change configuration of a topic named mytopic

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --add-config min.insync.replicas=1

3. Use the kafka-configs.sh tool to delete an existing configuration option.

   • Specify the host and port of the Kafka broker in the --bootstrap-server option.
   • Set the --entity-type as topic and --entity-name to the name of your topic.
   • Use --delete-config option to remove existing configuration option.

   • Specify the options you want to remove in the option --remove-config.

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_host>:<port> --entity-type topics --entity-name <topic_name> --alter --delete-config <option>

     Example of the command to change configuration of a topic named mytopic

     /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name mytopic --alter --delete-config min.insync.replicas

Deleting a topic

1. Delete a topic using the kafka-topics.sh utility.

   • Host and port of the Kafka broker in the --bootstrap-server option.
   • Use the --delete option to specify that an existing topic should be deleted.

   • Topic name must be specified in the --topic option.

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --delete --topic <topic_name>

     Example of the command to create a topic named mytopic

     /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic mytopic

2. Verify that the topic was deleted using kafka-topics.sh.

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --list

   Example of the command to list all topics

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

Procedure

1. Open the sample properties file in a text editor, or create a new one, and edit the file to include connection information and the replication flows for each Kafka cluster.

   The following example shows a configuration to connect two clusters, cluster-1 and cluster-2, bidirectionally. Cluster names are configurable through the clusters property.

   Example MirrorMaker 2 configuration

   clusters=cluster-1,cluster-2 1
   
   cluster-1.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_one>:443 2
   cluster-1.security.protocol=SSL 3
   cluster-1.ssl.truststore.password=<truststore_name>
   cluster-1.ssl.truststore.location=<path_to_truststore>/truststore.cluster-1.jks_
   cluster-1.ssl.keystore.password=<keystore_name>
   cluster-1.ssl.keystore.location=<path_to_keystore>/user.cluster-1.p12
   
   cluster-2.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_two>:443 4
   cluster-2.security.protocol=SSL 5
   cluster-2.ssl.truststore.password=<truststore_name>
   cluster-2.ssl.truststore.location=<path_to_truststore>/truststore.cluster-2.jks_
   cluster-2.ssl.keystore.password=<keystore_name>
   cluster-2.ssl.keystore.location=<path_to_keystore>/user.cluster-2.p12
   
   cluster-1->cluster-2.enabled=true 6
   cluster-2->cluster-1.enabled=true 7
   cluster-1->cluster-2.topics=.* 8
   cluster-2->cluster-1.topics=topic-1, topic-2 9
   cluster-1->cluster-2.groups=.* 10
   cluster-2->cluster-1.groups=group-1, group-2 11
   
   replication.policy.separator=- 12
   sync.topic.acls.enabled=false 13
   refresh.topics.interval.seconds=60 14
   refresh.groups.interval.seconds=60 15

   1

   Each Kafka cluster is identified with its alias.

   2

   Connection information for cluster-1, using the bootstrap address and port 443. Both clusters use port 443 to connect to Kafka using OpenShift Routes.

   3

   The ssl. properties define TLS configuration for cluster-1.

   4

   Connection information for cluster-2.

   5

   The ssl. properties define the TLS configuration for cluster-2.

   6

   Replication flow enabled from cluster-1 to cluster-2.

   7

   Replication flow enabled from cluster-2 to cluster-1.

   8

   Replication of all topics from cluster-1 to cluster-2. The source connector replicates the specified topics. The checkpoint connector tracks offsets for the specified topics.

   9

   Replication of specific topics from cluster-2 to cluster-1.

   10

   Replication of all consumer groups from cluster-1 to cluster-2. The checkpoint connector replicates the specified consumer groups.

   11

   Replication of specific consumer groups from cluster-2 to cluster-1.

   12

   Defines the separator used for the renaming of remote topics.

   13

   When enabled, ACLs are applied to synchronized topics. The default is false.

   14

   The period between checks for new topics to synchronize.

   15

   The period between checks for new consumer groups to synchronize.

2. OPTION: If required, add a policy that overrides the automatic renaming of remote topics. Instead of prepending the name with the name of the source cluster, the topic retains its original name.

   This optional setting is used for active/passive backups and data migration.

   replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy

3. OPTION: If you want to synchronize consumer group offsets, add configuration to enable and manage the synchronization:

   refresh.groups.interval.seconds=60
   sync.group.offsets.enabled=true 1
   sync.group.offsets.interval.seconds=60 2
   emit.checkpoints.interval.seconds=60 3

   1

   Optional setting to synchronize consumer group offsets, which is useful for recovery in an active/passive configuration. Synchronization is not enabled by default.

   2

   If the synchronization of consumer group offsets is enabled, you can adjust the frequency of the synchronization.

   3

   Adjusts the frequency of checks for offset tracking. If you change the frequency of offset synchronization, you might also need to adjust the frequency of these checks.

4. Start Kafka in the target clusters:

   /opt/kafka/bin/kafka-server-start.sh -daemon \
   /opt/kafka/config/kraft/server.properties

5. Start MirrorMaker with the cluster connection configuration and replication policies you defined in your properties file:

   /opt/kafka/bin/connect-mirror-maker.sh \
   /opt/kafka/config/connect-mirror-maker.properties

   MirrorMaker sets up connections between the clusters.

6. For each target cluster, verify that the topics are being replicated:

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --list

Procedure

1. Edit the MirrorMaker consumer.properties and producer.properties files to turn off MirrorMaker 2 features.

   For example:

   replication.policy.class=org.apache.kafka.mirror.LegacyReplicationPolicy 1
   
   refresh.topics.enabled=false 2
   refresh.groups.enabled=false
   emit.checkpoints.enabled=false
   emit.heartbeats.enabled=false
   sync.topic.configs.enabled=false
   sync.topic.acls.enabled=false

   1

   Emulate the previous version of MirrorMaker.

   2

   MirrorMaker 2 features disabled, including the internal checkpoint and heartbeat topics

2. Save the changes and restart MirrorMaker with the properties files you used with the previous version of MirrorMaker:

   su - kafka /opt/kafka/bin/kafka-mirror-maker.sh \
   --consumer.config /opt/kafka/config/consumer.properties \
   --producer.config /opt/kafka/config/producer.properties \
   --num.streams=2

   The consumer properties provide the configuration for the source cluster and the producer properties provide the target cluster configuration.

   MirrorMaker sets up connections between the clusters.

3. Start Kafka in the target cluster:

   su - kafka
   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

4. For the target cluster, verify that the topics are being replicated:

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_host>:<port> --list

Procedure

1. Switch to the kafka user:

   su - kafka

2. List all the broker loggers for a broker by using the kafka-configs.sh tool:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --describe --entity-type broker-loggers --entity-name BROKER-ID

   For example, for broker 0:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --describe --entity-type broker-loggers --entity-name 0

   This returns the logging level for each logger: TRACE, DEBUG, INFO, WARN, ERROR, or FATAL.

   For example:

   #...
   kafka.controller.ControllerChannelManager=INFO sensitive=false synonyms={}
   kafka.log.TimeIndex=INFO sensitive=false synonyms={}

3. Change the logging level for one or more broker loggers. Use the --alter and --add-config options and specify each logger and its level as a comma-separated list in double quotes.

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --alter --add-config "LOGGER-ONE=NEW-LEVEL,LOGGER-TWO=NEW-LEVEL" --entity-type broker-loggers --entity-name BROKER-ID

   For example, for broker 0:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server localhost:9092 --alter --add-config "kafka.controller.ControllerChannelManager=WARN,kafka.log.TimeIndex=WARN" --entity-type broker-loggers --entity-name 0

   If successful this returns:

   Completed updating config for broker: 0.

Procedure

1. Switch to the kafka user:

   su - kafka

2. Check the current logging level for the loggers configured in the connect-log4j.properties file:

   $ cat /opt/kafka/config/connect-log4j.properties
   
   # ...
   log4j.rootLogger=INFO, stdout, connectAppender
   # ...
   log4j.logger.org.reflections=ERROR

   Use a curl command to check the logging levels from the admin/loggers endpoint of the Kafka Connect API:

   curl -s http://localhost:8083/admin/loggers/ | jq
   
   {
     "org.reflections": {
       "level": "ERROR"
     },
     "root": {
       "level": "INFO"
     }
   }

   jq prints the output in JSON format. The list shows standard org and root level loggers, plus any specific loggers with modified logging levels.

   If you configure TLS (Transport Layer Security) authentication for the admin.listeners configuration in Kafka Connect, then the address of the loggers endpoint is the value specified for admin.listeners with the protocol as https, such as https://localhost:8083.

   You can also get the log level of a specific logger:

   curl -s http://localhost:8083/admin/loggers/org.apache.kafka.connect.mirror.MirrorCheckpointConnector | jq
   
   {
     "level": "INFO"
   }

3. Use a PUT method to change the log level for a logger:

   curl -Ss -X PUT -H 'Content-Type: application/json' -d '{"level": "TRACE"}' http://localhost:8083/admin/loggers/root
   
   {
     # ...
   
     "org.reflections": {
       "level": "TRACE"
     },
     "org.reflections.Reflections": {
       "level": "TRACE"
     },
     "root": {
       "level": "TRACE"
     }
   }

   If you change the root logger, the logging level for loggers that used the root logging level by default are also changed.

1. Reassign the partitions from the node you plan to remove to the remaining brokers using the Cruise control remove-nodes operation.
2. Update the cluster configuration, if necessary, to adjust the replication factor for topics (default.replication.factor) and the minimum required number of in-sync replica acknowledgements (min.insync.replicas).
3. Stop the Kafka broker service on the node and remove the node from the cluster.

Procedure

1. Unregister the removed node from the cluster:

   /opt/kafka/bin/kafka-cluster.sh unregister \
     --bootstrap-server <broker_host>:<port> \
     --id <node_id_number>

2. Verify the current state of the cluster by describing the topics:

   /opt/kafka/bin/kafka-topics.sh \
     --bootstrap-server <broker_host>:<port> \
     --describe

Procedure

1. Download the latest version of the Red Hat Streams for Apache Kafka Cruise Control archive from the Red Hat Customer Portal.

2. Create the /opt/cruise-control directory:

   sudo mkdir /opt/cruise-control

3. Extract the contents of the Cruise Control ZIP file to the new directory:

   unzip amq-streams-<version>-cruise-control-bin.zip -d /opt/cruise-control

4. Change the ownership of the /opt/cruise-control directory to the kafka user:

   sudo chown -R kafka:kafka /opt/cruise-control

1. Stop the Kafka broker:

   /opt/kafka/bin/kafka-server-stop.sh

2. Edit the Kafka configuration properties file to configure the Cruise Control Metrics Reporter.

   1. Add the CruiseControlMetricsReporter class to the metric.reporters configuration option. Do not remove any existing Metrics Reporters.

      metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter

   2. Add the following configuration options and values:

      cruise.control.metrics.topic.auto.create=true
      cruise.control.metrics.topic.num.partitions=1
      cruise.control.metrics.topic.replication.factor=1

      These options enable the Cruise Control Metrics Reporter to create the __CruiseControlMetrics topic with a log cleanup policy of DELETE. For more information, see Auto-created topics and Log cleanup policy for Cruise Control Metrics topic.

3. Configure SSL, if required.

   1. In the Kafka configuration properties file, configure SSL between the Cruise Control Metrics Reporter and the Kafka broker by setting the relevant client configuration properties.

      The Metrics Reporter accepts all standard producer-specific configuration properties with the cruise.control.metrics.reporter prefix. For example: cruise.control.metrics.reporter.ssl.truststore.password.

   2. In the Cruise Control properties file (/opt/cruise-control/config/cruisecontrol.properties) configure SSL between the Kafka broker and the Cruise Control server by setting the relevant client configuration properties.

      Cruise Control inherits SSL client property options from Kafka and uses those properties for all Cruise Control server clients.

4. Restart the Kafka broker:

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

   For information on restarting brokers in a multi-node cluster, see Section 3.6, “Performing a graceful rolling restart of Kafka brokers”.

5. Repeat steps 1-5 for the remaining brokers.

Procedure

1. Edit the Cruise Control properties file (/opt/cruise-control/config/cruisecontrol.properties).

2. Configure the properties shown in the following example configuration:

   # The Kafka cluster to control.
   bootstrap.servers=localhost:9092 1
   
   # The replication factor of Kafka metric sample store topic
   sample.store.topic.replication.factor=2 2
   
   # The configuration for the BrokerCapacityConfigFileResolver (supports JBOD, non-JBOD, and heterogeneous CPU core capacities)
   #capacity.config.file=config/capacity.json
   #capacity.config.file=config/capacityCores.json
   capacity.config.file=config/capacityJBOD.json 3
   
   # The list of goals to optimize the Kafka cluster for with pre-computed proposals
   default.goals={List of default optimization goals} 4
   
   # The list of supported goals
   goals={list of main optimization goals} 5
   
   # The list of supported hard goals
   hard.goals={List of hard goals} 6
   
   # How often should the cached proposal be expired and recalculated if necessary
   proposal.expiration.ms=60000 7
   
   #Set failure detection to true
   kafka.broker.failure.detection.enable=true 8

   1

   Host and port numbers of the Kafka broker (always port 9092).

   2

   Replication factor of the Kafka metric sample store topic. If you are evaluating Cruise Control in a single-node Kafka cluster, set this property to 1. For production use, set this property to 2 or more.

   3

   The configuration file that sets the maximum capacity limits for broker resources. Use the file that applies to your Kafka deployment configuration. For more information, see Capacity configuration.

   4

   Comma-separated list of default optimization goals, using fully-qualified domain names (FQDNs). A number of main optimization goals (see 5) are already set as default optimization goals; you can add or remove goals if desired. For more information, see Section 15.5, “Optimization goals overview”.

   5

   Comma-separated list of main optimization goals, using FQDNs. To completely exclude goals from being used to generate optimization proposals, remove them from the list. For more information, see Section 15.5, “Optimization goals overview”.

   6

   Comma-separated list of hard goals, using FQDNs. Seven of the main optimization goals are already set as hard goals; you can add or remove goals if desired. For more information, see Section 15.5, “Optimization goals overview”.

   7

   The interval, in milliseconds, for refreshing the cached optimization proposal that is generated from the default optimization goals. For more information, see Section 15.6, “Optimization proposals overview”.

   8

   Enables Cruise Control to use the Kafka API to detect broker failures.

3. Start the Cruise Control server. The server starts on port 9092 by default; optionally, specify a different port.

   cd /opt/cruise-control/
   ./kafka-cruise-control-start.sh config/cruisecontrol.properties <port_number>

4. To verify that Cruise Control is running, send a GET request to the /state endpoint of the Cruise Control server:

   curl -X GET 'http://<cc_host>:<cc_port>/kafkacruisecontrol/state'

1. Get the current configuration of the __CruiseControlMetrics topic:

   opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name __CruiseControlMetrics --describe

2. Change the log cleanup policy in the topic configuration:

   /opt/kafka/bin/kafka-configs.sh --bootstrap-server <broker_address> --entity-type topics --entity-name __CruiseControlMetrics --alter --add-config cleanup.policy=delete

Procedure

1. Generate an optimization proposal using a POST request to the /rebalance, /add_broker, or /remove_broker endpoint.

   Example request to /rebalance using default goals

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance'

   The cached optimization proposal is immediately returned.

   Note

   If NotEnoughValidWindows is returned, Cruise Control has not yet recorded enough metrics data to generate an optimization proposal. Wait a few minutes and then resend the request.

   Example request to /rebalance using specified goals

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?goals=RackAwareGoal,ReplicaCapacityGoal'

   If the request satisfies the supplied goals, the cached optimization proposal is immediately returned. Otherwise, a new optimization proposal is generated using the supplied goals; this takes longer to calculate. You can enforce this behavior by adding the ignore_proposal_cache=true parameter to the request.

   Example request to /rebalance using specified goals without hard goals

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?goals=RackAwareGoal,ReplicaCapacityGoal,ReplicaDistributionGoal&skip_hard_goal_check=true'

   Example request to /add_broker that includes specified brokers

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/add_broker?brokerid=3,4'

   The request includes the IDs of the new brokers only. For example, this request adds brokers with the IDs 3 and 4. Replicas are moved to the new brokers from existing brokers when rebalancing.

   Example request to /remove_broker that excludes specified brokers

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/remove_broker?brokerid=3,4'

   The request includes the IDs of the brokers being excluded only. For example, this request excludes brokers with the IDs 3 and 4. Replicas are moved from the brokers being removed to other existing brokers when rebalancing.

   Note

   If a broker that is being removed has excluded topics, replicas are still moved.

2. Review the optimization proposal contained in the response. The properties describe the pending cluster rebalance operation.

   The proposal contains a high level summary of the proposed optimization, followed by summaries for each default optimization goal, and the expected cluster state after the proposal has executed.

   Pay particular attention to the following information:

   • The Cluster load after rebalance summary. If it meets your requirements, you should assess the impact of the proposed changes using the high level summary.
   • n inter-broker replica (y MB) moves indicates how much data will be moved across the network between brokers. The higher the value, the greater the potential performance impact on the Kafka cluster during the rebalance.
   • n intra-broker replica (y MB) moves indicates how much data will be moved within the brokers themselves (between disks). The higher the value, the greater the potential performance impact on individual brokers (although less than that of n inter-broker replica (y MB) moves).
   • The number of leadership moves. This has a negligible impact on the performance of the cluster during the rebalance.

Procedure

1. Send a POST request to the /rebalance, /add_broker, or /remove_broker endpoint with the dryrun=false parameter:

   If you used the /add_broker or /remove_broker endpoint to generate a proposal that included or excluded brokers, use the same endpoint to perform the rebalance with or without the specified brokers.

   Example request to /rebalance

   curl -X POST 'cruise-control-server:9090/kafkacruisecontrol/rebalance?dryrun=false'

   Example request to /add_broker

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/add_broker?dryrun=false&brokerid=3,4'

   Example request to /remove_broker

   curl -v -X POST 'cruise-control-server:9090/kafkacruisecontrol/remove_broker?dryrun=false&brokerid=3,4'

   Cruise Control initiates the cluster rebalance and returns the optimization proposal.

2. Check the changes that are summarized in the optimization proposal. If the changes are not what you expect, you can stop the rebalance.

3. Check the progress of the cluster rebalance using the /user_tasks endpoint. The cluster rebalance in progress has a status of "Active".

   To view all cluster rebalance tasks executed on the Cruise Control server:

   curl 'cruise-control-server:9090/kafkacruisecontrol/user_tasks'
   
   USER TASK ID      CLIENT ADDRESS  START TIME     STATUS  REQUEST URL
   c459316f-9eb5-482f-9d2d-97b5a4cd294d  0:0:0:0:0:0:0:1       2020-06-01_16:10:29 UTC  Active      POST /kafkacruisecontrol/rebalance?dryrun=false
   445e2fc3-6531-4243-b0a6-36ef7c5059b4  0:0:0:0:0:0:0:1       2020-06-01_14:21:26 UTC  Completed   GET /kafkacruisecontrol/state?json=true
   05c37737-16d1-4e33-8e2b-800dee9f1b01  0:0:0:0:0:0:0:1       2020-06-01_14:36:11 UTC  Completed   GET /kafkacruisecontrol/state?json=true
   aebae987-985d-4871-8cfb-6134ecd504ab  0:0:0:0:0:0:0:1       2020-06-01_16:10:04 UTC

4. To view the status of a particular cluster rebalance task, supply the user-task-ids parameter and the task ID:

   curl 'cruise-control-server:9090/kafkacruisecontrol/user_tasks?user_task_ids=c459316f-9eb5-482f-9d2d-97b5a4cd294d'

1. Check that each broker being removed does not have any live partitions in its log (log.dirs).

   ls -l <LogDir> | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'

   If a log directory does not match the regular expression \.[a-z0-9]-delete$, active partitions are still present. If you have active partitions, check the rebalance has finished or the configuration for the optimization proposal. You can run the proposal again. Make sure that there are no active partitions before moving on to the next step.

2. Stop the broker.

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

3. Confirm that the broker has stopped.

   jcmd | grep kafka

Procedure

1. Create a configuration file for the new broker using the same settings as for the other brokers in your cluster, except for broker.id, which should be a number that is not already used by any of the other brokers.

2. Start the new Kafka broker passing the configuration file you created in the previous step as the argument to the kafka-server-start.sh script:

   su - kafka
   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

3. Verify that the Kafka broker is running.

   jcmd | grep Kafka

4. Repeat the above steps for each new broker.

5. If you haven’t done so, generate a reassignment JSON file named reassignment.json using the kafka-reassign-partitions.sh tool.

   Example command to generate the reassignment JSON file

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --topics-to-move-json-file topics.json \ 1
     --broker-list 0,1,2,3,4 \ 2
     --generate

   1

   The JSON file that specifies the topic.

   2

   Brokers IDs in the kafka cluster to include in the operation. This assumes broker 4 has been added.

   Example reassignment JSON file showing the current and proposed replica assignment

   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2,3],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3,4],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,4,0],"log_dirs":["any","any","any","any"]}]}

   Save a copy of this file locally in case you need to revert the changes later on.

6. Run the partition reassignment using the --execute option.

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --execute

   If you are going to throttle replication you can also pass the --throttle option with an inter-broker throttled rate in bytes per second. For example:

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --throttle 5000000 \
     --execute

7. Verify that the reassignment has completed using the --verify option.

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --verify

   The reassignment has finished when the --verify command reports that each of the partitions being moved has completed successfully. This final --verify will also have the effect of removing any reassignment throttles.

Procedure

1. If you haven’t done so, generate a reassignment JSON file named reassignment.json using the kafka-reassign-partitions.sh tool.

   Example command to generate the reassignment JSON file

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --topics-to-move-json-file topics.json \ 1
     --broker-list 0,1,2,3 \ 2
     --generate

   1

   The JSON file that specifies the topic.

   2

   Brokers IDs in the kafka cluster to include in the operation. This assumes broker 4 has been removed.

   Example reassignment JSON file showing the current and proposed replica assignment

   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[3,4,2,0],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[0,2,3,1],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[1,3,0,4],"log_dirs":["any","any","any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,0],"log_dirs":["any","any","any"]}]}

   Save a copy of this file locally in case you need to revert the changes later on.

2. Run the partition reassignment using the --execute option.

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --execute

   If you are going to throttle replication you can also pass the --throttle option with an inter-broker throttled rate in bytes per second. For example:

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --throttle 5000000 \
     --execute

3. Verify that the reassignment has completed using the --verify option.

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --verify

   The reassignment has finished when the --verify command reports that each of the partitions being moved has completed successfully. This final --verify will also have the effect of removing any reassignment throttles.

4. Check that each broker being removed does not have any live partitions in its log (log.dirs).

   ls -l <LogDir> | grep -E '^d' | grep -vE '[a-zA-Z0-9.-]+\.[a-z0-9]+-delete$'

   If a log directory does not match the regular expression \.[a-z0-9]-delete$, active partitions are still present. If you have active partitions, check the reassignment has finished or the configuration in the reassignment JSON file. You can run the reassignment again. Make sure that there are no active partitions before moving on to the next step.

5. Stop the broker.

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

6. Confirm that the Kafka broker has stopped.

   jcmd | grep kafka

Procedure

1. If you haven’t done so, generate a reassignment JSON file named reassignment.json using the kafka-reassign-partitions.sh tool.

   Example command to generate the reassignment JSON file

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --topics-to-move-json-file topics.json \ 1
     --broker-list 0,1,2,3,4 \ 2
     --generate

   1

   The JSON file that specifies the topic.

   2

   Brokers IDs in the kafka cluster to include in the operation.

   Example reassignment JSON file showing the current and proposed replica assignment

   Current partition replica assignment
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[3,4,2,0],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[0,2,3,1],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[1,3,0,4],"log_dirs":["any","any","any","any"]}]}
   
   Proposed partition reassignment configuration
   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2,3],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3,4],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,4,0],"log_dirs":["any","any","any","any"]}]}

   Save a copy of this file locally in case you need to revert the changes later on.

2. Edit the reassignment.json to remove a replica from each partition.

   For example use jq to remove the last replica in the list for each partition of the topic:

   Removing the last topic replica for each partition

   jq '.partitions[].replicas |= del(.[-1])' reassignment.json > reassignment.json

   Example reassignment file showing the updated replicas

   {"version":1,"partitions":[{"topic":"my-topic","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":1,"replicas":[1,2,3],"log_dirs":["any","any","any","any"]},{"topic":"my-topic","partition":2,"replicas":[2,3,4],"log_dirs":["any","any","any","any"]}]}

3. Make the topic replica change using the --execute option.

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --execute

   Note

   Removing replicas from a broker does not require any inter-broker data movement, so there is no need to throttle replication. If you are adding replicas, then you may want to change the throttle rate.

4. Verify that the change to the topic replicas has completed using the --verify option.

   /opt/kafka/bin/kafka-reassign-partitions.sh \
     --bootstrap-server localhost:9092 \
     --reassignment-json-file reassignment.json \
     --verify

   The reassignment has finished when the --verify command reports that each of the partitions being moved has completed successfully. This final --verify will also have the effect of removing any reassignment throttles.

5. Run the bin/kafka-topics.sh command with the --describe option to see the results of the change to the topics.

   /opt/kafka/bin/kafka-topics.sh \
     --bootstrap-server localhost:9092 \
     --describe

   Results of reducing the number of replicas for a topic

   my-topic  Partition: 0  Leader: 0  Replicas: 0,1,2 Isr: 0,1,2
   my-topic  Partition: 1  Leader: 2  Replicas: 1,2,3 Isr: 1,2,3
   my-topic  Partition: 2  Leader: 3  Replicas: 2,3,4 Isr: 2,3,4