Using AMQ Streams on RHEL

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>-bin.zip file into the /opt/kafka directory.

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

   This step requires admin privileges.

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

   chown -R kafka:kafka /opt/kafka

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

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

5. 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.

6. (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.0, and the Kafka Bridge.

Running the cluster

1. Edit the ZooKeeper configuration file /opt/kafka/config/zookeeper.properties. Set the dataDir option to /var/lib/zookeeper/:

   dataDir=/var/lib/zookeeper/

2. Edit the Kafka configuration file /opt/kafka/config/server.properties. Set the log.dirs option to /var/lib/kafka/:

   log.dirs=/var/lib/kafka/

3. Switch to the kafka user:

   su - kafka

4. Start ZooKeeper:

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

5. Check that ZooKeeper is running:

   jcmd | grep zookeeper

   Returns:

   number org.apache.zookeeper.server.quorum.QuorumPeerMain /opt/kafka/config/zookeeper.properties

6. Start Kafka:

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

7. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   number kafka.Kafka /opt/kafka/config/server.properties

Procedure

1. Start the Kafka console producer and configure it to send messages to a new topic:

   /opt/kafka/bin/kafka-console-producer.sh --broker-list <bootstrap_address> --topic <topic-name>

   For example:

   /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic

2. Enter several messages into the console. Press Enter to send each individual message to your new topic:

   >message 1
   >message 2
   >message 3
   >message 4

   When Kafka creates a new topic automatically, you might receive a warning that the topic does not exist:

   WARN Error while fetching metadata with correlation id 39 :
   {4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

   The warning should not reappear after you send further messages.

3. In a new terminal window, start the Kafka console consumer and configure it to read messages from the beginning of your new topic.

   /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server <bootstrap_address> --topic <topic-name> --from-beginning

   For example:

   /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

   The incoming messages display in the consumer console.

4. Switch to the producer console and send additional messages. Check that they display in the consumer console.
5. Stop the Kafka console producer and then the consumer by pressing Ctrl+C.

Procedure

1. Stop the Kafka broker.

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

2. Confirm that the Kafka broker is stopped.

   jcmd | grep kafka

3. Stop ZooKeeper.

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

Running the cluster

1. Create the myid file in /var/lib/zookeeper/. Enter ID 1 for the first ZooKeeper node, 2 for the second ZooKeeper node, and so on.

   su - kafka
   echo "<NodeID>" > /var/lib/zookeeper/myid

   For example:

   su - kafka
   echo "1" > /var/lib/zookeeper/myid

2. Edit the ZooKeeper /opt/kafka/config/zookeeper.properties configuration file for the following:

   • Set the option dataDir to /var/lib/zookeeper/.
   • Configure the initLimit and syncLimit options.
   • Configure the reconfigEnabled and standaloneEnabled options.

   • Add a list of all ZooKeeper nodes. The list should include also the current node.

     Example configuration for a node of ZooKeeper cluster with five members

     tickTime=2000
     dataDir=/var/lib/zookeeper/
     initLimit=5
     syncLimit=2
     reconfigEnabled=true
     standaloneEnabled=false
     listener.security.protocol.map=PLAINTEXT:PLAINTEXT,REPLICATION:PLAINTEXT
     
     server.1=172.17.0.1:2888:3888:participant;172.17.0.1:2181
     server.2=172.17.0.2:2888:3888:participant;172.17.0.2:2181
     server.3=172.17.0.3:2888:3888:participant;172.17.0.3:2181
     server.4=172.17.0.4:2888:3888:participant;172.17.0.4:2181
     server.5=172.17.0.5:2888:3888:participant;172.17.0.5:2181

3. Start ZooKeeper with the default configuration file.

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

4. Verify that ZooKeeper is running.

   jcmd | grep zookeeper

   Returns:

   number org.apache.zookeeper.server.quorum.QuorumPeerMain /opt/kafka/config/zookeeper.properties

5. Repeat this procedure on all the nodes of the cluster.

6. Verify that all nodes are members of the cluster by sending a stat command to each of the nodes using the ncat utility.

   Use ncat stat to check the node status

   echo stat | ncat localhost 2181

   To use four-letter word commands, like stat, you need to specify 4lw.commands.whitelist=* in zookeeper.properties.

   The output shows that a node is either a leader or follower.

   Example output from the ncat command

   ZooKeeper version: 3.4.13-2d71af4dbe22557fda74f9a9b4309b15a7487f03, built on 06/29/2018 00:39 GMT
   Clients:
    /0:0:0:0:0:0:0:1:59726[0](queued=0,recved=1,sent=0)
   
   Latency min/avg/max: 0/0/0
   Received: 2
   Sent: 1
   Connections: 1
   Outstanding: 0
   Zxid: 0x200000000
   Mode: follower
   Node count: 4

1. Edit the /opt/kafka/config/server.properties Kafka configuration file as follows:

   • Set the broker.id field to 0 for the first broker, 1 for the second broker, and so on.
   • Configure the details for connecting to ZooKeeper in the zookeeper.connect option.
   • Configure the Kafka listeners.

   • Set the directories where the commit logs should be stored in the logs.dir directory.

     Here we see an example configuration for a Kafka broker:

     broker.id=0
     zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181
     listeners=REPLICATION://:9091,PLAINTEXT://:9092
     listener.security.protocol.map=PLAINTEXT:PLAINTEXT,REPLICATION:PLAINTEXT
     inter.broker.listener.name=REPLICATION
     log.dirs=/var/lib/kafka

     In a typical installation where each Kafka broker is running on identical hardware, only the broker.id configuration property will differ between each broker config.

2. Start the Kafka broker with the default configuration file.

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

3. Verify that the Kafka broker is running.

   jcmd | grep Kafka

   Returns:

   number kafka.Kafka /opt/kafka/config/server.properties

4. Verify that all nodes are members of the Kafka cluster by sending a dump command to one of the ZooKeeper nodes using the ncat utility.

   Use ncat dump to check all Kafka brokers registered in ZooKeeper

   echo dump | ncat zoo1.my-domain.com 2181

   To use four-letter word commands, like dump, you need to specify 4lw.commands.whitelist=* in zookeeper.properties.

   The output must contain all Kafka brokers you just configured and started.

   Example output from the ncat command for a Kafka cluster with 3 nodes

   SessionTracker dump:
   org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9
   ephemeral nodes dump:
   Sessions with Ephemerals (3):
   0x20000015dd00000:
           /brokers/ids/1
   0x10000015dc70000:
           /controller
           /brokers/ids/0
   0x10000015dc70001:
           /brokers/ids/2

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 brokers and ZooKeeper

3. Restart the Kafka broker:

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

4. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   number kafka.Kafka /opt/kafka/config/server.properties

5. Verify that all nodes are members of the Kafka cluster by sending a dump command to one of the ZooKeeper nodes using the ncat utility.

   Use ncat dump to check all Kafka brokers registered in ZooKeeper

   echo dump | ncat zoo1.my-domain.com 2181

   To use four-letter word commands, like dump, you need to specify 4lw.commands.whitelist=* in zookeeper.properties.

   The output must contain the Kafka broker you started.

   Example output from the ncat command for a Kafka cluster with 3 nodes

   SessionTracker dump:
   org.apache.zookeeper.server.quorum.LearnerSessionTracker@28848ab9
   ephemeral nodes dump:
   Sessions with Ephemerals (3):
   0x20000015dd00000:
           /brokers/ids/1
   0x10000015dc70000:
           /controller
           /brokers/ids/0
   0x10000015dc70001:
           /brokers/ids/2

6. 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 <bootstrap_address>  --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. Generate an ID for the Kafka cluster using the kafka-storage tool:

   /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 broker 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, process.roles=broker, controller.

   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.

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.

   The default log directory location specified in the server.properties configuration file is /tmp/kraft-combined-logs. You can add a comma-separated list to set up multiple log directories.

4. Start each Kafka broker.

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

5. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   number kafka.Kafka /opt/kafka/config/kraft/server.properties

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
   config.providers.env.class=io.strimzi.kafka.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.3, “Performing a graceful rolling restart of Kafka brokers”.

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 will describe all available topics.

     /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

     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_address> --entity-type topics --entity-name <TopicName> --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_address> --entity-type topics --entity-name <TopicName> --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_address> --entity-type topics --entity-name <TopicName> --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_address> --delete --topic <TopicName>

     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_address> --list

   Example of the command to list all topics

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

Procedure

1. Create the /opt/kafka/connector-plugins directory.

   su - kafka
   mkdir /opt/kafka/connector-plugins

2. Edit the /opt/kafka/config/connect-standalone.properties or /opt/kafka/config/connect-distributed.properties Kafka Connect configuration file, and set the plugin.path option to /opt/kafka/connector-plugins. For example:

   plugin.path=/opt/kafka/connector-plugins

3. Copy your connector plug-ins to /opt/kafka/connector-plugins.
4. Start or restart the Kafka Connect workers.

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.0 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 ZooKeeper and Kafka in the target clusters:

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

   /opt/kafka/bin/kafka-server-start.sh -daemon \
   /opt/kafka/config/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_address> --list

Procedure

1. Edit the MirrorMaker consumer.properties and producer.properties files to turn off MirrorMaker 2.0 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.0 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 ZooKeeper and Kafka in the target cluster:

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

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

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

   /opt/kafka/bin/kafka-topics.sh --bootstrap-server <broker_address> --list

Procedure

1. Edit the /opt/kafka/config/server.properties Kafka configuration file.

   The plugin properties are shown in this example configuration.

   Example Kafka Static Quota plugin configuration

   # ...
   client.quota.callback.class=io.strimzi.kafka.quotas.StaticQuotaCallback 1
   client.quota.callback.static.produce=1000000 2
   client.quota.callback.static.fetch=1000000 3
   client.quota.callback.static.storage.soft=400000000000 4
   client.quota.callback.static.storage.hard=500000000000 5
   client.quota.callback.static.storage.check-interval=5 6
   # ...

   1

   Loads the Kafka Static Quota plugin.

   2

   Sets the producer byte-rate threshold. 1 MBps in this example.

   3

   Sets the consumer byte-rate threshold. 1 MBps in this example.

   4

   Sets the lower soft limit for storage. 400 GB in this example.

   5

   Sets the higher hard limit for storage. 500 GB in this example.

   6

   Sets the interval in seconds between checks on storage. 5 seconds in this example. You can set this to 0 to disable the check.

2. Start the Kafka broker with the default configuration file.

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

3. Verify that the Kafka broker is running.

   jcmd | grep Kafka

Procedure

1. Add the Red Hat Maven repository to the <repositories> section of your pom.xml file.

   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="http://maven.apache.org/POM/4.0.0"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   
       <!-- ... -->
   
       <repositories>
           <repository>
               <id>redhat-maven</id>
               <url>https://maven.repository.redhat.com/ga/</url>
           </repository>
       </repositories>
   
       <!-- ... -->
   
   </project>

2. Add the clients to the <dependencies> section of your pom.xml file.

   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="http://maven.apache.org/POM/4.0.0"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   
       <!-- ... -->
   
       <dependencies>
           <dependency>
               <groupId>org.apache.kafka</groupId>
               <artifactId>kafka-clients</artifactId>
               <version>3.3.1.redhat-00008</version>
           </dependency>
       </dependencies>
   
       <!-- ... -->
   </project>

3. Build your Maven project.

Procedure

1. Add the Red Hat Maven repository to the <repositories> section of your pom.xml file.

   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="http://maven.apache.org/POM/4.0.0"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   
       <!-- ... -->
   
       <repositories>
           <repository>
               <id>redhat-maven</id>
               <url>https://maven.repository.redhat.com/ga/</url>
           </repository>
       </repositories>
   
       <!-- ... -->
   
   </project>

2. Add kafka-streams to the <dependencies> section of your pom.xml file.

   <?xml version="1.0" encoding="UTF-8"?>
   <project xmlns="http://maven.apache.org/POM/4.0.0"
            xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
            xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
   
       <!-- ... -->
   
       <dependencies>
           <dependency>
               <groupId>org.apache.kafka</groupId>
               <artifactId>kafka-streams</artifactId>
               <version>3.3.1.redhat-00008</version>
           </dependency>
       </dependencies>
   
       <!-- ... -->
   </project>

3. Build your Maven project.

1. Service principals
2. Kafka brokers to use the Kerberos login
3. ZooKeeper to use Kerberos login
4. Producer and consumer clients to access Kafka using Kerberos authentication

1. Create the service principals, and keytabs that store the principal keys, through the Kerberos KDC.

   For example:

   • zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM
   • producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

   • consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM

     The ZooKeeper service principal must have the same hostname as the zookeeper.connect configuration in the Kafka config/server.properties file:

     zookeeper.connect=node1.example.redhat.com:2181

     If the hostname is not the same, localhost is used and authentication will fail.

2. Create a directory on the host and add the keytab files:

   For example:

   /opt/kafka/krb5/zookeeper-node1.keytab
   /opt/kafka/krb5/kafka-node1.keytab
   /opt/kafka/krb5/kafka-producer1.keytab
   /opt/kafka/krb5/kafka-consumer1.keytab

3. Ensure the kafka user can access the directory:

   chown kafka:kafka -R /opt/kafka/krb5

1. Create or modify the opt/kafka/config/jaas.conf file to support ZooKeeper client and server operations:

   Client {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true 1
       storeKey=true 2
       useTicketCache=false 3
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab" 4
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM"; 5
   };
   
   Server {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumServer {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   
   QuorumLearner {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/zookeeper-node1.keytab"
       principal="zookeeper/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

   1

   Set to true to get the principal key from the keytab.

   2

   Set to true to store the principal key.

   3

   Set to true to obtain the Ticket Granting Ticket (TGT) from the ticket cache.

   4

   The keyTab property points to the location of the keytab file copied from the Kerberos KDC. The location and file must be readable by the kafka user.

   5

   The principal property is configured to match the fully-qualified principal name created on the KDC host, which follows the format SERVICE-NAME/FULLY-QUALIFIED-HOST-NAME@DOMAIN-NAME.

2. Edit opt/kafka/config/zookeeper.properties to use the updated JAAS configuration:

   # ...
   
   requireClientAuthScheme=sasl
   jaasLoginRenew=3600000 1
   kerberos.removeHostFromPrincipal=false 2
   kerberos.removeRealmFromPrincipal=false 3
   quorum.auth.enableSasl=true 4
   quorum.auth.learnerRequireSasl=true 5
   quorum.auth.serverRequireSasl=true
   quorum.auth.learner.loginContext=QuorumLearner 6
   quorum.auth.server.loginContext=QuorumServer
   quorum.auth.kerberos.servicePrincipal=zookeeper/_HOST 7
   quorum.cnxn.threads.size=20

   1

   Controls the frequency for login renewal in milliseconds, which can be adjusted to suit ticket renewal intervals. Default is one hour.

   2

   Dictates whether the hostname is used as part of the login principal name. If using a single keytab for all nodes in the cluster, this is set to true. However, it is recommended to generate a separate keytab and fully-qualified principal for each broker host for troubleshooting.

   3

   Controls whether the realm name is stripped from the principal name for Kerberos negotiations. It is recommended that this setting is set as false.

   4

   Enables SASL authentication mechanisms for the ZooKeeper server and client.

   5

   The RequireSasl properties controls whether SASL authentication is required for quorum events, such as master elections.

   6

   The loginContext properties identify the name of the login context in the JAAS configuration used for authentication configuration of the specified component. The loginContext names correspond to the names of the relevant sections in the opt/kafka/config/jaas.conf file.

   7

   Controls the naming convention to be used to form the principal name used for identification. The placeholder _HOST is automatically resolved to the hostnames defined by the server.1 properties at runtime.

3. Start ZooKeeper with JVM parameters to specify the Kerberos login configuration:

   su - kafka
   export EXTRA_ARGS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   If you are not using the default service name (zookeeper), add the name using the -Dzookeeper.sasl.client.username=NAME parameter.

   Note

   If you are using the /etc/krb5.conf location, you do not need to specify -Djava.security.krb5.conf=/etc/krb5.conf when starting ZooKeeper, Kafka, or the Kafka producer and consumer.

1. Modify the opt/kafka/config/jaas.conf file with the following elements:

   KafkaServer {
       com.sun.security.auth.module.Krb5LoginModule required
       useKeyTab=true
       storeKey=true
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };
   KafkaClient {
       com.sun.security.auth.module.Krb5LoginModule required debug=true
       useKeyTab=true
       storeKey=true
       useTicketCache=false
       keyTab="/opt/kafka/krb5/kafka-node1.keytab"
       principal="kafka/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   };

2. Configure each broker in the Kafka cluster by modifying the listener configuration in the config/server.properties file so the listeners use the SASL/GSSAPI login.

   Add the SASL protocol to the map of security protocols for the listener, and remove any unwanted protocols.

   For example:

   # ...
   broker.id=0
   # ...
   listeners=SECURE://:9092,REPLICATION://:9094 1
   inter.broker.listener.name=REPLICATION
   # ...
   listener.security.protocol.map=SECURE:SASL_PLAINTEXT,REPLICATION:SASL_PLAINTEXT 2
   # ..
   sasl.enabled.mechanisms=GSSAPI 3
   sasl.mechanism.inter.broker.protocol=GSSAPI 4
   sasl.kerberos.service.name=kafka 5
   ...

   1

   Two listeners are configured: a secure listener for general-purpose communications with clients (supporting TLS for communications), and a replication listener for inter-broker communications.

   2

   For TLS-enabled listeners, the protocol name is SASL_PLAINTEXT. For non-TLS-enabled connectors, the protocol name is SASL_PLAINTEXT. If SSL is not required, you can remove the ssl.* properties.

   3

   SASL mechanism for Kerberos authentication is GSSAPI.

   4

   Kerberos authentication for inter-broker communication.

   5

   The name of the service used for authentication requests is specified to distinguish it from other services that may also be using the same Kerberos configuration.

3. Start the Kafka broker, with JVM parameters to specify the Kerberos login configuration:

   su - kafka
   export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/opt/kafka/config/jaas.conf"; /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

   If the broker and ZooKeeper cluster were previously configured and working with a non-Kerberos-based authentication system, it is possible to start the ZooKeeper and broker cluster and check for configuration errors in the logs.

   After starting the broker and Zookeeper instances, the cluster is now configured for Kerberos authentication.

1. Add the Kerberos configuration to the producer or consumer configuration file.

   For example:

   /opt/kafka/config/producer.properties

   # ...
   sasl.mechanism=GSSAPI 1
   security.protocol=SASL_PLAINTEXT 2
   sasl.kerberos.service.name=kafka 3
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \ 4
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/producer1.keytab" \
       principal="producer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

   1

   Configuration for Kerberos (GSSAPI) authentication.

   2

   Kerberos uses the SASL plaintext (username/password) security protocol.

   3

   The service principal (user) for Kafka that was configured in the Kerberos KDC.

   4

   Configuration for the JAAS using the same properties defined in jaas.conf.

   /opt/kafka/config/consumer.properties

   # ...
   sasl.mechanism=GSSAPI
   security.protocol=SASL_PLAINTEXT
   sasl.kerberos.service.name=kafka
   sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
       useKeyTab=true  \
       useTicketCache=false \
       storeKey=true  \
       keyTab="/opt/kafka/krb5/consumer1.keytab" \
       principal="consumer1/node1.example.redhat.com@EXAMPLE.REDHAT.COM";
   # ...

2. Run the clients to verify that you can send and receive messages from the Kafka brokers.

   Producer client:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-producer.sh --producer.config /opt/kafka/config/producer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094

   Consumer client:

   export KAFKA_HEAP_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Dsun.security.krb5.debug=true"; /opt/kafka/bin/kafka-console-consumer.sh --consumer.config /opt/kafka/config/consumer.properties  --topic topic1 --bootstrap-server node1.example.redhat.com:9094