Introduction to the Kafka Demo
Hi guys, welcome to Code Decode. Today in this video, we will see how to implement Kafka in real time with a live demo. And then understanding this kind of architecture for Kafka will be much more easier after we know how this actually is internally working. So let's go and create a live demo first of how we can create producers, how we can create consumers, how to create end-to-end configuration for Kafka, and then let's understand how this actually works internally. So let's get started.
Please like, share, and subscribe to support us. And we are setting a like target of 500 likes.
Prerequisites: Installing Zookeeper & Kafka
For implementation of Kafka, the very first thing we need is we need Zookeeper and Kafka installed for us. Now, since we are running all this in local and we are not using Docker at all, we will be installing all of these things in the local first. And then later on, when we are good with the basic demo for the Kafka, the producers and consumers, we will move to Docker later on after this much is clear to us.
So for installing the Kafka and Zookeeper, the link is here, and this will also be given in the description below. Just go ahead and install Kafka from this particular link, Scala 2.12, and you will have an inbuilt Zookeeper already installed with the Kafka. Now I have this already installed. So I've installed it in the software package. If you can see, in the bin, you can see with Kafka you have Zookeeper already installed.
So once you have done this particular step of installing the Zookeeper and Kafka just by downloading it from here and extracting it at some folder location, you are good to go with the start of the Spring Boot project creation.
Designing the Producer-Consumer Architecture
For Kafka, the project setup for Kafka is we need a Spring Boot project, then we need a Kafka dependency with the web dependency. Then we need a controller to fetch the message from a client. So we are going to create an architecture where there will be a producer, where there will be a consumer, and in between will be a Kafka topic. Topics are also called as brokers, and brokers have internal partitions.
So we need to produce something, and that producing message I'll be taking from the REST API from the client. So there will be a third-party client who wants to send some data to the consumer. So he'll hit the REST API. The REST API will contact producer and give the message to the producer. Producer will produce that message in the topic of the broker, and that will be subscribed by the consumer, and consumer will pick it. So what we need is as a wholesome then, we need a Spring Boot project to have the producer, consumer, and Kafka into it.
Secondly, in the Spring Boot project, we need a REST controller which will act as an interface between the client and the producer. So the task of that REST API will be to take the message from the client and send it to producer. Producer will send that message to the broker, and then we need a consumer who will subscribe to that broker. So first thing, a REST API. Second, we need a producer. Third, we need a consumer. And fourth, we need the configuration settings for the implementation.
So you need a controller REST API, you need a producer who pushes the message to the topic, you need a consumer to listen and subscribe to it. Next, you need the properties file to have the configuration for the Kafka. At last, when everything is done, you need to run first the Zookeeper, then you need to run the Kafka server, and then if you try to hit something, you will be able to send that message from producer to consumer. So let's get started with a quick demo.
Creating the Spring Boot Project
Let's quickly create a new Spring Starter project. I can name it as Kafka Demo. So this is a demo project for Kafka implementation. Let's go to next and let's have the Spring Web as the dependency and Kafka as a dependency.
We are not using streams. After a few videos, when we are good with the streams, we'll also create a video on streams and how to create streams and what all APIs they have added for the streams. So currently, we have Kafka and Web, and let's finish it. So we have Kafka Demo project almost ready with this. This is the main class which will run.
Building the REST Controller Endpoint
Now, what we need is a controller. So we'll create a package for controller and a class RestEndpointForKafka. Here, this is a REST controller, so we will be annotating it with @RestController. Here, we need a GET mapping, getMessageFromClient. It will have a @RequestParameter, say it as message, and it will be of type String.
We'll have a request mapping here which will identify this controller uniquely. So it's an API for REST. So this is basically a REST API. In the GET mapping, we'll have a unique name as producer-message. So this message is going to be a message for producer. So whatever you give as a request parameter here will be a message for the producer to send to the topic.
Implementing the Kafka Producer Service
Now our controller is ready, but we need a producer to produce the message to the topic. This is a task with some kind of business logic into it, right? So we need a service package. That service package will have two services: a producer service and consumer service. And similarly, we will have a consumer service. Now producer service will be annotated with @Service because a bean should be created for this. And similarly, consumer should also be a service.
With this in place, we need to send something to Kafka, right? So we need a KafkaTemplate. And it's going to have two parameters. First parameter will be the name of the topic, and second will be the message. So the name of the topic will be, for sure, be of type String, and similarly the value will be also of type String. And I'll name it as kafkaTemplate. and I'm going to autowire it.
With this, we have a simple method named as public void sendMessageToTopic and it will have a String message as a parameter. And simple, the kafkaTemplate will be used to send the message to the topic. So it will have two parameters: the name of the topic. So you have to first decide the name of the topic. So let's make the name of the topic as code_decode_topic. Fine. And let's send the message to the topic. So I can say my producer is ready. My producer is also a bean. It has autowired a KafkaTemplate, and KafkaTemplate is capable enough to send my message to the newly created topic. Great.
Connecting the Controller to the Producer
Now the producer is ready. So in the controller what we can do is we can auto wire this producer and you can use this producer to send message to the topic with the message that you will receive from the client.
So this is going to be hit by the client. It will give you a message in the request parameter itself as a GET request. You are going to get that message and send it to the topic through the producer which we have recently created. So this producer is going to call the kafkaTemplate.send method, which will internally send this message that you are sending as a parameter to the topic created with this name. Now this is all good, we have the producer ready. We have our controller ready, but we need a consumer also, right?
Implementing the Kafka Consumer Service
So consumer will be something who will listen, listen to your topic. So there will be a method, public void listenToTopic, and it will have a String message which it will read or consume from the topic. So this will be a received message, and I'm just simply going to print it. Great.
But this method has to be annotated with something which will make it capable enough to read or listen to the messages from a topic, or which will make this class capable enough to be a subscriber to a topic. So we will have an annotation called as @KafkaListener. This @KafkaListener is going to make this class or this method capable enough to be a Kafka listener or a consumer to the Kafka topic. But it needs two things, that is topic. So basically you can have multiple topics to be listened to, so this is basically an array of topics you can listen to. I have one single topic here. What topic we have? This is code_decode_topic which we have created. So single topic is going to be listened and subscribed to.
And a groupId. This is very important because this Kafka is made to be fault-tolerant and scalable. Now if you have hundreds of consumers listening to a single topic, it is for scalability internally divided into multiple partitions. Now each partition is nothing but a replica of some data to reduce the load from one single partition. The task of this groupId is to group these consumers into groups so that the load can be balanced in these broker. Not C1, C2, and C3 should not go and read from just first partition and bombard this first partition with multiple requests to read. Rather, C1 should read from first partition, C2 should read from second, C3 from the third, and so on. To make it load balanced, the consumers are grouped with the groupId. So if you have C1, C2 and C3 in the same group, say code_decode_group, then the data will be load balanced between C1 and C2 so that you do not bombard with the pull request for the data on this topic. Each consumer will read from a separate partition, and these particular partitions are equally balanced between all the consumers in a particular group. So this is why this groupId is very important. The groupId here is going to be, say suppose, code_decode_group.
Now our Kafka listener is ready, our Kafka producer is ready, our REST endpoint is ready.
Configuring Kafka Properties in application.yaml
When all these things are ready, we need configuration for Kafka implementation, right? So in the main/resources, we will be creating a new file named as application.yaml. You can also do that in a properties file, but for the simplicity, I use the YAML file. At 9092, this is for the Kafka default server port. Now, whenever we send a message from a producer through a template to a topic, it has to be serialized. That is, the string has to be converted into the type of bytes. Now how to serialize the message and the value, the keys and the values to send it to the particular topic is through serialization and deserialization. So I'm denoting that both the type of key and value serialization should be of string serialization because I want to convert a string to type of byte so that it can be sent to the topic to the consumer at the end. These are nothing but the type of serializations we are using for conversion of your message to bytes so that it can be moved from producer to consumer through the topic.
Great, we have the Kafka configuration ready, we have the producer ready, we have the consumer ready, we have the controller ready. Let's, what are we waiting for? Let's run it and let's see, some exception will for sure happen. Why? Because we have not started the Zookeeper and the Kafka on the local system.
Starting the Zookeeper and Kafka Servers
So see, we get an exception. Broker, bootstrap broker 9092, it is disconnected. Cannot establish the connection. So how do you establish a connection? Go here and open the CMD here. I'm going to give you two commands. First is to run Zookeeper, second is to run Kafka. Both will be given in the description below.
For Zookeeper, this is the command that at bin/windows, start the batch for Zookeeper server and get the properties from the config folder's zookeeper.properties file. This will run the Zookeeper for us now. And after that, we should run one more command to run the Kafka. Now, this is the command to run the Kafka. So in the same folder, if you can go here, you can see in the bin folder, you have a zookeeper-start-server.bat and also the kafka-server-start.bat. So we are starting the Kafka server and we are also starting the Zookeeper server. So you can see this was the zookeeper-start-server. Similarly, we need the Kafka server to be started and also get the properties for the server from this properties file.
The Kafka is running for us and it was able to find a group has one member. So let's go and see. We were facing issue of connection error could not be established, could not be established. As soon as we started our replica Zookeeper and Kafka, it says successfully joined the group. And the group name is something which we have created: code-decode-group. It says successfully synced, adding the assigned partition to the topic. The topic name which we have created was code-decode-topic, right? What was the topic name? code-decode-topic. Can you see that code-decode-topic here? And the partition is also added for that. This proves that our Kafka is now up and running. We have successfully started the Kafka server and the Zookeeper server.
Live Demo: End-to-End Message Flow
Now let's try and hit this controller with the REST API producer-message. So it should be localhost:8080 because we are running on local, /rest/api/producer-message. /rest/api/producer-message and the request parameter should be message. And that message should be congrats for first kafka impl and hit. Congress for the first kafka empl. So our console successfully says the message is received.
This message is received by the consumer who is going to listen to this topic. The topic I have already shown you in the previous console that the topic was ready. So what happened here was the producer was called with this particular REST API. You hit this REST API producer-message with this message, congrats. So this congrats message comes here. The producer is called to send this message to the topic. When you go to the producer sendMessageToTopic method, a KafkaTemplate is used to send this message, congrats for the first kafka implementation, to code_decode_topic. In the console in the previous seconds, we have seen this topic was ready with one partition as zeroth index.
As soon as the topic has the message, it has a listener available. The listener was this consumer who was a @KafkaListener for this topic in this code-decode group that I have already shown in the console also. It listened to that particular received message and it prints the message is received, that is, the received message is this. So this was the whole flow of this particular project. This project is already been uploaded in this particular git location: github/codedecode25/kafka. SRC and everything is available here, you can take a reference from here. While creating, if you find any issues, you can always refer to this particular public repository for which we have created for you.
Recap and What's Next
Now with all these things in place, you have all these steps that you have done. You have created a Kafka dependency in POM, you have created a controller, you have a producer which is a service who is going to publish to the topic, you have a consumer who is going to listen to a topic. You have created some property changes, all of these property changes will be present here in application.properties file, in this main/resources you can find this application.yaml file.
Then, going to the location we have unzipped, where in the first step we have unzipped Kafka, you have started the Zookeeper server and the Kafka server. And then you hit with this particular message. Now what all we have done is, we must have the Zookeeper for Kafka. Zookeeper performs many tasks for Kafka, but basically Zookeeper manages the Kafka cluster. By default Kafka service is started at 9092 and the Zookeeper started at 2181. That I'll show in the next part where these configurations you can see. That we have already seen a config folder, there all these configurations are there.
Now, groupId I already told you, consumers in Kafka are divided into multiple groups and each group is identified uniquely by the groupId. The consumers belong to the same group, share the groupId. So the topic partitions are load balanced fairly between them. The consumers in a group divide the partitions fairly among themselves, as fairly as possible, to establish that each partition is only consumed by a single consumer from the group. So that they will not hit multiple consumers at a time to a single partition and bombard and heavily load that partition. Not to do that, each partition is going to be consumed by only a single consumer from that single group.
In KafkaTemplate, I've already told you that is used to send a message. We have used KafkaTemplate here. The send method is used. The first parameter is a topic name, second is the message that you want to send to that topic. So all you need to do is the send method is called with the topic name and message. In order to prepare the message for transmission from the producer to the broker, we use serialization. I have already told you why we use serialization and deserialization, because we need to send data in bytes in the topic. So you have to convert your string to bytes and then deserialize bytes to string so as to receive that from the topic. That is why you need the StringSerializer that we have already created as a configuration part in the properties file.
Now we have many more questions: how to change the Zookeeper default port number 2181 and how to change the Kafka port number that is 9092, and what are the terminologies and architecture of Kafka? What was Zookeeper, why is it actually used? How is a cluster divided into multiple partitions? And many, many more things. And for that, we'll be meeting in the next video. Just let me know in the comment section if we can continue with the series. Thank you.
