Apache Kafka
MULTI-NODE
CLUSTER CREATION
This
guide covers the installation of a multi-node cluster only. Be aware that this setup is not compatible with Cloudera, Hortonworks or other vendors' clusters and it’s based on the official Apache Kafka documentation and steps for standalone clusters.
See also: Kafka - How to create a single-node Apache Kafka cluster
See also: Kafka - How to create a single-node Apache Kafka cluster
1. Download
Kafka
a. Log in via SSH to the first node that will be part
of the Kafka cluster, switch to root user and download the Kafka tar file.
$ sudo -i
$ mkdir /tmp/kafka
$ cd /tmp/kafka
b. Extract the tar file, create the required directories
and copy the extracted files within, then apply specific permissions on them:
$ tar -xzf
kafka_2.11-0.10.2.0.tgz
$ mkdir /usr/lib/kafka
$ chown -R root:root
/usr/lib/kafka
$ rsync -av
kafka_2.11-0.10.2.0/* /usr/lib/kafka/
Move to the /usr/lib/kafka directory and create a
symlink between the config subdirectory and /etc/kafka/conf
$ mkdir
/etc/kafka
$ cd
/etc/kafka
$ ln -s
/usr/lib/kafka/config /etc/kafka/conf
Create a directory for Kafka in /var/log and a
symlink between /usr/lib/kafka/logs and /var/log/kafka/logs:
$ mkdir
/var/log/kafka/
$
chmod -R 774 /var/log/kafka/
$ chown -R root:root
/var/log/kafka/
$ cd /var/log/kafka
$ ln
-s /usr/lib/kafka/logs /var/log/kafka/logs
c. Repeat the same process for the rest of the
cluster’s nodes.
2. Configure
Kafka
In
the below example, we will configure 3 broker nodes. Note that each broker will
exist in a separate node as we are configuring a multi-node Kafka cluster.
a. Log in to the first node of the cluster
b. Look for the extra mounted disk (not covered in this guide) that will be used to
store Kafka data. In the below example, the disk is mounted in /DATA
$ df -h
Create the Kafka’s data and Zookeeper directories,
and apply permissions as below:
$ mkdir -p /DATA/kafka/data
$ chown -R root:root
/DATA/kafka
$ chmod -R 775 /DATA/kafka
$ mkdir -p
/DATA/kafka/zookeeper
$ chown -R root:root /DATA/kafka/zookeeper
$ chmod -R 775 /DATA/kafka/zookeeper
c. Create a backup of the original server.properties
file:
$ cp
/etc/kafka/conf/server.properties /etc/kafka/conf/server.properties.bkp
d. Create a backup of the original zookeeper.properties
file
$ cp /etc/kafka/conf/zookeeper.properties
/etc/kafka/conf/zookeeper.properties.bkp
e. Repeat the above steps for all the nodes in the
cluster.
f. Edit the server.properties file and add/modify
each line as shown below. Make sure to replace the lines that already exist
instead of adding them twice. We will make these changes in all the nodes in
the cluster:
Node 1:
$ vi /etc/kafka/conf/server.properties
broker.id=1
listeners=PLAINTEXT://:9092
log.dirs=/DATA/kafka/data
zookeeper.connect=<Node1_FQDN>:2181,<Node2_FQDN>:2181,<Node3_FQDN>:2181
Repeat the same for the rest of the nodes in the
cluster. Make sure to increment the “broker.id” and the port in “listeners” as
shown below.
Node 2:
$ vi /etc/kafka/conf/server.properties
broker.id=2
listeners=PLAINTEXT://:9093
log.dirs=/DATA/kafka/data
zookeeper.connect=<Node1_FQDN>:2181,<Node2_FQDN>:2181,<Node3_FQDN>:2181
Node 3:
$ vi /etc/kafka/conf/server.properties
broker.id=3
listeners=PLAINTEXT://:9094
log.dirs=/DATA/kafka/data
zookeeper.connect=<Node1_FQDN>:2181,<Node2_FQDN>:2181,<Node3_FQDN>:2181
Make sure to replace <Node#_FQDN> by
the actual FQDN of each Kafka server.
g. Edit the zookeeper.properties file and
add/modify each line as shown below. Make sure to replace the lines that
already exist instead of adding them twice. We will make these changes in
all the nodes in the cluster:
Node 1:
$ vi /etc/kafka/conf/zookeeper.properties
dataDir=/DATA/kafka/zookeeper
clientPort=2181
maxClientCnxns=0
server.1=<Node1_FQDN>:2888:3888
server.2=<Node1_FQDN>:2888:3888
server.3=<Node1_FQDN>:2888:3888
initLimit=5
syncLimit=2
Node 2:
$ vi /etc/kafka/conf/zookeeper.properties
dataDir=/DATA/kafka/zookeeper
clientPort=2181
maxClientCnxns=0
server.1=<Node1_FQDN>:2888:3888
server.2=<Node1_FQDN>:2888:3888
server.3=<Node1_FQDN>:2888:3888
initLimit=5
syncLimit=2
Node 3:
$ vi /etc/kafka/conf/zookeeper.properties
dataDir=/DATA/kafka/zookeeper
clientPort=2181
maxClientCnxns=0
server.1=<Node1_FQDN>:2888:3888
server.2=<Node1_FQDN>:2888:3888
server.3=<Node1_FQDN>:2888:3888
initLimit=5
syncLimit=2
h. Create a “myid” file in the zookeeper directory as
shown below. We will make these changes in all the nodes in the cluster, setting
the myid number accordingly:
Node 1:
$ echo "1" >
/DATA/kafka/zookeeper/myid
$ sudo chown root:root /DATA/kafka/zookeeper/myid
$ sudo chmod 644 /DATA/kafka/zookeeper/myid
Node 2:
$ echo "2" >
/DATA/kafka/zookeeper/myid
$ sudo chown root:root /DATA/kafka/zookeeper/myid
$ sudo chmod 644 /DATA/kafka/zookeeper/myid
Node 3:
$ echo "3" >
/DATA/kafka/zookeeper/myid
$ sudo chown root:root /DATA/kafka/zookeeper/myid
$ sudo chmod 644 /DATA/kafka/zookeeper/myid
i. Modify the permissions of the start and stop
scripts so that no one other than root can execute
them. We will make these changes in all the nodes in the cluster:
Node 1, 2 and
3:
$ cd /usr/lib/kafka/bin
$ chmod 774 zookeeper-server-start.sh
zookeeper-server-stop.sh
$ chmod 774 kafka-server-start.sh
kafka-server-stop.sh
3. Start
and Test Kafka
To start Kafka, we first
need to start the embedded Zookeeper daemons in all the nodes, which
will act as coordinators for the different Kafka brokers and their
corresponding jobs.
Once the Zookeeper daemons
are started, we can proceed to start the Kafka servers (a.k.a. brokers) as per
our previous configuration.
- To start all the above-mentioned daemons in the background, we will add the “-daemon” argument right before specifying the configuration files (where the daemons get the required settings to start).
- To start the daemons in the foreground, no arguments must be passed to the script, other than including the configuration files (as mentioned above).
Since we’re starting the
daemons for the first time, we will execute them in the foreground
(using several different terminals) to easily visualize any error that may
occur.
a. Log in to each Kafka node, then start the Zookeeper
daemon in the foreground as root. Make sure to pay attention to the
command’s output in case of any error:
Node
1:
$ sudo
/usr/lib/kafka/bin/zookeeper-server-start.sh /etc/kafka/conf/zookeeper.properties
Node
2:
$ sudo
/usr/lib/kafka/bin/zookeeper-server-start.sh
/etc/kafka/conf/zookeeper.properties
Node
3:
$ sudo
/usr/lib/kafka/bin/zookeeper-server-start.sh
/etc/kafka/conf/zookeeper.properties
b. With separate terminals, log in to each Kafka node
again and start each Kafka broker (configured as per the server.properties
file on each node) in the foreground as root:
Node
1:
$ sudo /usr/lib/kafka/bin/kafka-server-start.sh
/usr/lib/kafka/config/server.properties
Node
2:
$ sudo /usr/lib/kafka/bin/kafka-server-start.sh
/usr/lib/kafka/config/server.properties
Node 3:
$ sudo /usr/lib/kafka/bin/kafka-server-start.sh
/usr/lib/kafka/config/server.properties
c. We now have Zookeeper running in all the nodes, as
well as 3 instances of Kafka broker, one per node.
The next step will be to create a test topic with a replication factor of
3 (consistent with the number of brokers previously created). Make sure to
replace <Node#_FQDN> by the actual FQDN of each of the Kafka nodes.
Open a new terminal
connected to any node in the cluster and run the following command (in this example we’re using the Node 1):
Node 1:
$ sudo
/usr/lib/kafka/bin/kafka-topics.sh --create --zookeeper <Node1_FQDN>:2181,<Node2_FQDN>:2181,<Node3_FQDN>:2181
--replication-factor 3 --partitions 1 --topic testtopic
You can validate the topic was created by running
the below script:
$
/usr/lib/kafka/bin/kafka-topics.sh --describe --zookeeper <Node1_FQDN>:2181,<Node2_FQDN>:2181,<Node3_FQDN>:2181
--topic testtopic
d. Once the topic is created and validated, we can test
the following scenarios:
·
Sending
messages from a Kafka producer
·
Visualizing
those messages in a Kafka consumer
·
Testing
failover between the 3 brokers (i.e. force-stopping brokers to validate that
the remaining ones keep on sending/receiving messages)
In a different node, execute a Kafka consumer
pointing to our testtopic, which will listen for incoming messages from
a Kafka producer (we will use Node 3 in this example):
Node
3:
$ /usr/lib/kafka/bin/kafka-console-consumer.sh
--bootstrap-server <Node1_FQDN>:9092, <Node2_FQDN>:9093, <Node3_FQDN>:9094
--topic testtopic --from-beginning
Open another terminal, connect to a different node and
execute a Kafka producer pointing to the same topic (in this example we’re
using the Node 2):
Node
2:
$
/usr/lib/kafka/bin/kafka-console-producer.sh --broker-list <Node1_FQDN>:9092,
<Node2_FQDN>:9093, <Node3_FQDN>:9094 --topic testtopic
Once the producer in Node 2 is started, type
any message you want and press <ENTER> to submit it. You can send as much
messages as you want.
Go back to the Node 3 and make sure that all
messages are being received in the Kafka consumer console.
See thee below examples for clarification (disregard
the number of brokers showed after “--broker-list”):
Producer terminal (sending messages)
Consumer terminal (receiving messages)
e. Open yet another terminal to force-stop a broker and
test a failover scenario (in this example we will stop Kafka in Node 1):
Node
1:
Look for the process associated with the server.properties
file and stop it.
$ ps -ef | grep server.properties
(…)
$ kill -9 12831
Once stopped, you should still be able to send and
receive messages with the Kafka producer and consumer (if there’s at least one
broker up). Repeat the process described in step d to validate
that you can still send and receive messages.
f. f. Once you’ve validated that the failover scenario
works, stop all the running kafka processes, including Zookeeper. The below
script will stop all the processes associated with Kafka brokers:
Node
1, 2 and 3:
$ sudo
/usr/lib/kafka/bin/kafka-server-stop.sh
Wait a few seconds and then stop the Zookeeper process:
$ sudo
/usr/lib/kafka/bin/zookeeper-server-stop.sh
Validate that no Kafka/Zookeeper daemons are still running:
$ ps -ef | grep kafka
g. In the next step, we will start Zookeeper and Kafka
as daemons so that they can run in the background and don’t depend on any
opened user session.
4. Start/Stop
Zookeeper and Kafka daemons
In this section, we will
finally start Zookeeper and Kafka as background daemons so that the cluster is
operational. We will do so using the root user
a. Start Zookeeper and Kafka brokers. Make sure to
start them in order as shown below:
Node 1, 2 and
3:
$ sudo
/usr/lib/kafka/bin/zookeeper-server-start.sh -daemon
/usr/lib/kafka/config/zookeeper.properties
Wait a few seconds and then start the Kafka brokers:
Node
1, 2, and 3:
$ sudo
/usr/lib/kafka/bin/kafka-server-start.sh -daemon /usr/lib/kafka/config/server.properties
Once all is started, you can repeat the steps described in 3.d to make sure that everything works as expected.
b. In case you need to, here’s how you can stop Kafka
brokers and then Zookeeper. Make sure to stop them in order as shown below:
$ sudo
/usr/lib/kafka/bin/kafka-server-stop.sh
Wait a few seconds and then stop Zookeeper:
$ sudo /usr/lib/kafka/bin/zookeeper-server-stop.sh
