Creating a Kafka Topic Running Kafka Brokers
Learn how to run Kafka topics using Kafka brokers in this article by Raúl Estrada, a programmer since 1996 and a Java developer since 2001. He has been an enterprise architect for BEA Systems and Oracle Inc., but he also enjoys web, mobile, and game programming. Raúl is a supporter of free software and enjoys experimenting with new technologies, frameworks, languages, and methods.
The real art behind a server is in its configuration. This article will examine how to deal with the basic configuration of a Kafka broker in standalone mode. There are two types of configuration: standalone and cluster. The real power of Kafka is unlocked when running with replication in cluster mode and all topics are correctly partitioned.
The cluster mode has two main advantages: parallelism and redundancy. Parallelism is the capacity to run tasks simultaneously among the cluster members. The redundancy warrants that when a Kafka node goes down, the cluster is safe and accessible from the other running nodes.
This article shows how to configure a cluster with several nodes on our local machine although, in practice, it is always better to have several machines with multiple nodes sharing clusters.
To start this exercise, let’s use Confluent Platform open source version. Confluent Platform is available also in Docker images, but here we are going to install it in local. Open Confluent Platform download page at https://www.confluent.io/download/ .
The current version of Confluent Platform is 5.0.0 as a stable release. Remember that, since the Kafka core runs on Scala, there are two versions: for Scala 2.11 and Scala 2.12.
We could run Confluent Platform from our desktop directory, but let’s use /opt/ for Linux users and /usr/local for macOS users.
To install Confluent Platform, extract the downloaded file, confluent-5.0.0-2.11.tar.gz, in the directory, as follows:
> tar xzf confluent-5.0.0-2.11.tar.gzNow, go to the Confluent Platform installation directory, referenced from now on as <confluent-path>.
A broker is a server instance. A server (or broker) is actually a process running in the operating system and starts based on its configuration file.
The people of Confluent have provided us with a template of a standard broker configuration. This file, which is called server.properties, is located in the Kafka installation directory in the config subdirectory:
1. Inside <confluent-path>, make a directory with the name mark.
2. For each Kafka broker (server) that we want to run, we need to make a copy of the configuration file template and rename it accordingly. In this example, our cluster is going to be called mark:
> cp config/server.properties <confluent-path>/mark/mark-1.properties> cp config/server.properties <confluent-path>/mark/mark-2.properties
3. Modify each properties file accordingly. If the file is called mark-1, the broker.id should be 1. Then, specify the port in which the server will run; the recommendation is 9093 for mark-1 and 9094 for mark-2. Note that the port property is not set in the template, so add the line. Finally, specify the location of the Kafka logs (a Kafka log is a specific archive to store all of the Kafka broker operations); in this case, use the /tmp directory. Here, it is common to have problems with write permissions. Do not forget to give write and execute permissions to the user with whom these processes are executed over the log directory, as in the examples:
· In mark-1.properties, set the following:
broker.id=1port=9093log.dirs=/tmp/mark-1-logs
· In mark-2.properties, set the following:
broker.id=2port=9094log.dirs=/tmp/mark-2-logs
4. Start the Kafka brokers using the kafka-server-start command with the corresponding configuration file passed as the parameter. Don't forget that Confluent Platform must be already running and the ports should not be in use by another process. Start the Kafka brokers as follows:
> <confluent-path>/bin/kafka-server-start <confluent-path>/mark/mark-1.properties &
And, in another command-line window, run the following command:
> <confluent-path>/bin/kafka-server-start <confluent-path>/mark/mark-2.properties &
Don’t forget that the trailing & is to specify that you want your command line back. If you want to see the broker output, it is recommended to run each command separately in its own command-line window.
Remember that the properties file contains the server configuration and that the server.properties file located in the config directory is just a template.
Now there are two brokers, mark-1 and mark-2 , running in the same machine in the same cluster.
Remember, there are no dumb questions, as in the following examples:
Q: How does each broker know which cluster it belongs to?
A: The brokers know that they belong to the same cluster because, in the configuration, both point to the same Zookeeper cluster.
Q: How does each broker differ from the others within the same cluster?
A: Every broker is identified inside the cluster by the name specified in the broker.id property.
Q: What happens if the port number is not specified?
A: If the port property is not specified, Zookeeper will assign the same port number and will overwrite the data.
Q: What happens if the log directory is not specified?
A: If log.dir is not specified, all the brokers will write to the same default log.dir. If the brokers are planned to run in different machines, then the port and log.dir properties might not be specified (because they run in the same port and log file but in different machines).
Q: How can I check that there is not a process already running in the port where I want to start my broker?
A: There is a useful command to see what process is running on specific port; in this case the 9093 port:
> lsof -i :9093The output of the previous command is something like this:
COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAMEjava 12529 admin 406u IPv6 0xc41a24baa4fedb11 0t0 TCP *:9093 (LISTEN)
Your turn: try to run this command before starting the Kafka brokers, and run it after starting them to see the change. Also, try to start a broker on a port in use to see how it fails.
OK, what if I want my cluster to run on several machines?
To run Kafka nodes on different machines but in the same cluster, adjust the Zookeeper connection string in the configuration file; its default value is as follows:
zookeeper.connect=localhost:2181Remember that the machines must be able to be found by each other by DNS and that there are no network security restrictions between them.
The default value for Zookeeper connect is correct only if you are running the Kafka broker in the same machine as Zookeeper. Depending on the architecture, it will be necessary to decide if there will be a broker running on the same Zookeeper machine.
To specify that Zookeeper might run in other machines, do the following:
zookeeper.connect=localhost:2181, 192.168.0.2:2183, 192.168.0.3:2182The previous line specifies that Zookeeper is running in the local host machine on port 2181, in the machine with IP address 192.168.0.2 on port 2183 , and in the machine with IP address, the 192.168.0.3, on port 2182. The Zookeeper default port is 2181, so normally it runs there.
Your turn: as an exercise, try to start a broker with incorrect information about the Zookeeper cluster. Also, using the lsof command, try to raise Zookeeper on a port in use.
If you have doubts about the configuration, or it is not clear what values to change, the server.properties template (as all of the Kafka project) is open sourced in the following:
https://github.com/apache/kafka/blob/trunk/config/server.properties
Running Kafka topics
The power inside a broker is the topic, namely the queues inside it. Now that we have two brokers running, let’s create a Kafka topic on them.
Kafka, like almost all modern infrastructure projects, has three ways of building things: through the command line, through programming, and through a web console (in this case the Confluent Control Center). The management (creation, modification, and destruction) of Kafka brokers can be done through programs written in most modern programming languages. If the language is not supported, it could be managed through the Kafka REST API. The previous section showed how to build a broker using the command line.
Is it possible to only manage (create, modify, or destroy) brokers through programming? No, we can also manage the topics. The topics can also be created through the command line. Kafka has pre-built utilities to manage brokers as we already saw and to manage topics, as we will see next.
To create a topic called amazingTopic in our running cluster, use the following command:
> <confluent-path>/bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic amazingTopicThe output should be as follows:
Created topic amazingTopicHere, the kafka-topics command is used. With the --create parameter it is specified that we want to create a new topic. The --topic parameter sets the name of the topic, in this case, amazingTopic.
Do you remember the terms parallelism and redundancy? Well, the –-partitions parameter controls the parallelism and the --replication-factor parameter controls the redundancy.
The --replication-factor parameter is fundamental as it specifies in how many servers of the cluster the topic is going to replicate (for example, running). On the other hand, one broker can run just one replica.
Obviously, if a greater number than the number of running servers on the cluster is specified, it will result in an error (you don’t believe me? Try it in your environment). The error will be like this:
Error while executing topic command: replication factor: 3 larger than available brokers: 2[2018-09-01 07:13:31,350] ERROR org.apache.kafka.common.errors.InvalidReplicationFactorException: replication factor: 3 larger than available brokers: 2(kafka.admin.TopicCommand$)
To be considered, the broker should be running (don’t be shy and test all this theory in your environment).
The --partitions parameter, as its name implies, says how many partitions the topic will have. The number determines the parallelism that can be achieved on the consumer's side. This parameter is very important when doing cluster fine-tuning.
Finally, as expected, the --zookeeper parameter indicates where the Zookeeper cluster is running.
When a topic is created, the output in the broker log is something like this:
[2018-09-01 07:05:53,910] INFO [ReplicaFetcherManager on broker 1] Removed fetcher for partitions amazingTopic-0 (kafka.server.ReplicaFetcherManager)[2018-09-01 07:05:53,950] INFO Completed load of log amazingTopic-0 with 1 log segments and log end offset 0 in 21 ms (kafka.log.Log)
In short, this message reads like a new topic has been born in our cluster.
How can I check my new and shiny topic? Use the same command: kafka-topics.
There are more parameters than --create. To check the status of a topic, run the kafka-topics command with the --list parameter, as follows:
> <confluent-path>/bin/kafka-topics.sh --list --zookeeper localhost:2181The output is the list of topics, as we know, is as follows:
amazingTopicThis command returns the list with the names of all of the running topics in the cluster.
How can I get details of a topic? Use the same command: kafka-topics.
For a particular topic, run the kafka-topics command with the --describe parameter, as follows:
> <confluent-path>/bin/kafka-topics --describe --zookeeper localhost:2181 --topic amazingTopicThe command output is as follows:
Topic:amazingTopic PartitionCount:1 ReplicationFactor:1 Configs: Topic: amazingTopic Partition: 0 Leader: 1 Replicas: 1 Isr: 1Here is a brief explanation of the output:
· PartitionCount: Number of partitions on the topic (parallelism)
· ReplicationFactor: Number of replicas on the topic (redundancy)
· Leader: Node responsible for reading and writing operations of a given partition
· Replicas: List of brokers replicating this topic data; some of these might even be dead
· Isr: List of nodes that are currently in-sync replicas
Let’s create a topic with multiple replicas (for example, we will run with more brokers in the cluster); we type the following:
> <confluent-path>/bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 2 --partitions 1 --topic redundantTopicThe output is as follows:
Created topic redundantTopicNow, call the kafka-topics command with the --describe parameter to check the topic details, as follows:
> <confluent-path>/bin/kafka-topics --describe --zookeeper localhost:2181 --topic redundantTopicTopic:redundantTopic PartitionCount:1 ReplicationFactor:2 Configs:Topic: redundantTopic Partition: 0 Leader: 1 Replicas: 1,2 Isr: 1,2
As you can see, Replicas and Isr are the same lists; we infer that all of the nodes are in-sync.
Your turn: play with the kafka-topics command, and try to create replicated topics on dead brokers and see the output. Also, create topics on running servers and then kill them to see the results. Was the output what you expected?
All these commands executed through the command line can be executed programmatically or performed through the Confluent Control Center web console.
If you found this article interesting, you can explore Apache Kafka Quick Start Guide to process large volumes of data in real-time while building high performance and robust data stream processing pipeline using the latest Apache Kafka 2.0. Apache Kafka Quick Start Guide will help you learn how to use Apache Kafka for efficient processing of distributed applications and will get familiar with solving everyday problems in fast data and processing pipelines.
