Kafka Producer and Consumer with Java: Hands-On Tutorial

Key Takeaways

hey guys welcome to the new video this is going to be a quick Hands-On video on Kafka producers and consumers we will not cover Kafka fundamentals in this video because we have already covered them if you are interested please refer this video in this video we will focus on a simple exercise to interact with Kafka producers and consumers so let's get [Music] started so let's see what we are going to cover in this video here we have a simple exercise on Kafka producer and consumer I will share the link of this repository so you can go through the details on your own now in this scenario we have a topic named user signups this is going to be a Kafka topic then the task is to develop a producer a simple Kafka producer that will generate the events basically user sign up event and it will send the events to this particular topic now this user sign up event will have a set of details like user ID username and time stamp so that means we will have to kind of manage a custom event all right it's not a simple primitive value like string or integer we have to send a custom event so we will also see how to create a custom event and how to work with Kafka with a custom event now the second step is basically to develop the consumer which will consume these events from the same topic which is user signups it will consume the topics and then it will print some information on the logs as an exercise so suppose we we have Kafka up and running on our system it could be a Docker container it could be a cloud offering or whatever it is we have a Kafka cluster up and running then we will need a component which is called producer and this producer could be a simple Java program I script A micros service or anything which is generating some events and any entity which is generating an event and is sending that event to Kafka is called a Kafka producer similarly we will have another component which is called consumer in this sense because it will read those events it will consume those events from the CFA although for another scenario for another set of events it can be a producer as well because it not only is consuming events from the Kafka but maybe in some other case it is also generating a set of events so a consumer can be a producer or a consumer depending on the activity all right let me revert the direction so we have the producer and the consumer these are some applications which are producing the events sending it to Kafka then consuming those events from the Kafka now these producers and consumers are also called Kafka clients because they both are clients of the Kafka cluster so when we have the producer who is interested in sending the events how does it know where to send the events for that we need something called topic so we create a topic and then the producer will send the new events to this topic like this and the consumer will consume the events from this topic now the second step is how does a client know where this Kafka is running for this we need to configure something called bootstrap server which is a property which will be passed to the consumer and the producer basically any application which is interacting with this Kafka cluster so we will set bootstrap servers we will provide this config the next step is to develop the producer and the consumer and for that we will use Kafka dependencies and from those dependencies we have classes like Kafka producer which we will use to basically develop the producer and similarly we have another class which is called Kafka consumer okay so we will use these classes in the demo to develop the producer and the consumer now when the producer sends an event to the topic what it actually does it sends something called producer record and this producer record encapsulates the actual event something like this okay so when the producer is sending an event it is actually sending an object of producer record and similarly when the consumer consumes the event instead of the Raw event what it consumes basically is something called consumer record and this consumer record internally holds that particular event which will be extracted by the client and the last thing to understand is when we have the event when we send an event to the Kafka you generally pass a key and the value all right now what Kafka does it actually converts this event to the the bite form okay so it needs to understand how to convert the event to the bite form and for that we need something called serializer and deserializer okay and these are also called seray in short form we don't always create the new Sur days for each and every type because there are well-known types like string and integer for which these serializers and deserializers already exist in the Kafka distribution so we can simply use them but for custom key and custom values for custom objects we have to develop the serializers and deserializers as well because Kafka does not understand the custom types we have to tell Kafka how to translate a custom type so in this example because we have a custom user event which is user signup event you will see how do we write a custom serializer and Des serializer and how do we pass that information to Kafka so let's start with the demo now first of all let's run the Kafka and create the topic so I will use Docker desktop to run the Kafka container and you can refer the last video in which we learned how to run Kafka using Docker desktop all right so we will go to the repository and I will copy the command and before that I will run a new Kafka container for this demo so I will go to my images and for this particular Kafka image I will run a new container and I'm going to name it Kafka exercise one and I will bind the same port 9092 and then I will start a new container so the Kafka container is up next step is to create the topic for that we will go to the exact tab which is same as running Docker exact command because we need to access the terminal and here we will go to the opt directory and in this one Kafka bin and here we have all the executables all the scripts required to run the kaf commands in order to create the topic I will copy the command that I created and in this one we have the name of the topic which is user signups bootstrap server because the Kafka container is running on local and the same port that we provided which is 9092 that's all let me create the topic and we have a new topic which is called user signups the next step is to write the code for producers and consumers so for that we will go to the intellig so the first step is to add the required dependencies we will not use spring boot or or any other abstractions it will be a simple Kafka and Java application so let's add the required dependencies and the main dependency that we need is Kafka clients this one which is coming from Kafka and we will use the latest version at the time of recording which is 3.9.0 that's it let me reload the for.xml and in the meantime we will also add another dependency which is required for custom serializer and der serial izer because we need to we will use something called object mapper basically to map the pojo or the object to bite form and this particular dependency is Jackson data bind this one and this is coming from Jackson core all right and the version that we are going to use is 2.8.1 and similarly we need another dependency for the Jackson and I will explain why but let me add the dependency for now now and this is Jackson data type jsr 310 coming from the same group and using the same version so the dependencies have been downloaded let's move on in the next step we will create the event so let's start by adding the new event user sign up event and from the description we know that we have to support couple of fields in this event which is going to be user ID username and sign up time for that we will use local date time like this then I will add a default Constructor and the parameterized Constructor okay I will also add Getters and Setters like this we can also use longbo for this one for example but I'm keeping it simple just the Kafka and simple Java libraries now because we have this custom event we also need to add custom serializer and deserializer otherwise Kafka won't understand how to convert this object to the bite form in order to send it across the network so let's start by adding the serializer and der serializer next so we will name it user signup event serializer so in order to write the serializer we need to implement an interface which is serializer and this comes from Kafka itself this one and we need to pass the type parameter which is going to be user sign up event in this case all right let's implement the Missing Method and we need to implement serialized method here the first argument or first parameter of this method is basically the name of the topic and the second parameter is the event that it will convert to the bite array here it is now to simplify the translation process we will use something called object mapper that's why we added the Jackson data band dependency object mapper and uh let's initialize the object mapper like this and we need to perform an additional step here which is object mapper doind register modules the reason is because in this event there is a field which is local date time which it cannot convert by default so for that we need additional dependency and this is the one that we added jsr 310 and to use that dependency we need to call this method find and register modules which will register basically that jsr module coming from Jackson datab bind and then it will be able to convert that local datetime field all right now from this particular method in order to convert it to bite array we will simply use object mapper do write value and we will pass the event that's it we can handle it basically the TR catch and for now this is plain and simple we are simply throwing the exception but in a real scenario we will properly handle the exception okay so for this we have the serializer ready and in the same way we will write the Der serializer like this and here we will do the same thing we need to implement an interface which is Der serializer coming from Kafka we will again pass the typ parameter which is user sign up event and uh implement the deserialize method like this and let me copy this part because we need to do the same thing with the Der serializer again and from this this method we will use object mapper do read value so the first parameter is the bite array because we need to convert this bite array and the second parameter is the object type basically that we are returning from here and in this case it will be user signup event. class and what this der serializer will do from this bite array it will basically create the object of user signup event which was sent from the uh prod sir that is the purpose of serialization and der serialization so we have the event ready and we have the serializer and der serializer ready all right let's move on so let's write the Kafka producer now we will add a new class and we will call it simple producer and in the simple producer we will have a main method because this is going to be a console application so as we learned we use Kafka producer class to create Kafka producers basically and so we will create an object of Kafka producer this is coming from the Kafka itself and it accepts two type parameters for key and value now in this demo the key would be integer we will use an integer key basically and the value would be the event itself which is user sign up event Kafka producer now let's instantiate this Kafka producer in the main method so we will have simple producer like this and let's write a method where we will instantiate this gafka producer now in this method as we talked about when we started the video that we need something called bootstrap server because otherwise the producer or the client does not know where this CFA is running then we also need to provide the serializers and deserializers for key and value okay so we will provide all these configurations in this method so what we need is we create something called a set of properties and by using this set of properties we pass the required configuration dop put the first thing is we need to provide the bootstrap server so there is this class producer config which has the constants for all these properties that we are going to use and here we can find that one which is bootstrap server config and this is the name of that particular property that Kafka understands which is bootstrap do servers but we are going to use the constant here all right the value would be Local Host colon 9092 because that's where the Kafka is running in this case that's where we executed or ran the Kafka container all right now the second conf that we need is props dop put producer config do key serializer class config which is basically the key serializer key. serializer so we are going to use this property and in this case because we are going to use integer and int value as a key so we can use an existing serializer we don't need to write a new serializer for this one and that existing serializer is if we search for the class this is something called integer serializer and if you notice this is coming from Kafka hyphen clients so we can basically copy the package path from here like this do integer serializer that's it and third property that we need to set here is reducer config do value serializer because we also need to provide the serial ier for the value now value in this case is the custom event that we created so we need to pass the custom serializer that we developed in this case and now this is going to be this serializer that we created so let me copy the package path like this and the name of that serializer so I will copy the name as well and that's it so that is the bare minimum that we need at this stage basically where the Kafka is running and how to serialize the event the key and the value itself then when we instantiate the Kafka producer when we call the Kafka producers Constructor we pass the configurations the set of properties to the Constructor and that's it so that is how we create a new object of Kafka producer now once we have the Kafka producer what we will do in this demo we will have an inter active console application that will take some inputs from the user and then it will generate the event accordingly it has nothing to do with CA it's just to make it interactive so we will use the same scanner class to basically take the input from the user like this okay so the first thing it will do it will ask for the uh ID the user ID basically now ideally we will generate the user ID automatically but in this demo application we will take this as an input so we will say enter the user ID and similarly we will also take the username as an input and this will be simple next and string username okay so we have the ID we have the name the next thing is basically to create the object of user signup event okay and send it to Kafka we will write another method which will be simple producer dot generate event and here we will create the new object of user sign up event so the first parameter is ID that we received from the user similarly we will also pass the username the third argument was the sign up time so we can pass the current time by calling local datetime.now method okay so let's create this method in the same class and in this method we will use the Kafka producer that we created we will call the send method that's how we send the event to Kafka so you can see this send method basically accepts couple of parameters like producer record producer record and call back so we will use this particular method where we will basically send the object of producer record and this producer record will internally hold the access to the event that we generated so let's create the object of producer record now this producer record accepts few parameters the first one is the name of the topic so in this case the name of the topic is going to be user signups because that is the topic that we created the second parameter is the key and as we discussed the key is going to be the integer type and in this case we can actually use the user ID because that is also integer so what I will do I will get the key like this okay and the third parameter is the payload itself which is going to be the event itself like this and then we can add message event sent all right now because this is going to be uh an infinite Loop so we need to terminate the loop so what we can do here we can ask the user whether to continue or to terminate so we will say type c to continue okay and Q2 wait it like this SC do next and here we will say if the input is quit Okay then we need to quit the program exiting and what we can do we can actually here call the methods to clean up the producer because in the CFA producer class we have methods like shutdown so if you notice this is a very simple program and we are not dealing with the exceptions we are not closing or releasing the connections and so on but in the real project we will do that so this is the place where you can let's say call CA producer. shut down basically to properly and gracefully shut down the producer and in the case of a console application we can also register a shutdown hook where we will close the producer but let's keep it simple so we will simply say return or you can say system. exit whatever it is so what we did in the simple producer we are actually creating the object of kka producer by passing the required configurations okay then in the main method we are taking two input values from the user user ID user name then we are creating the object of the custom event sending it to cfab in the form of producer record so this will now send the event to Kafka now let's write the consumer to read those events okay so we will create a new class here simple consumer and this will also have a main method and in this main method we will create the object of kakka consumer here also we need to provide the same type parameters for keys and values so we know the key is basically integer and the value is the event which is user signup event afka consumer and we will write a similar method for the consumer also which is start consumer so let's write this method and here also we need to provide the required configurations so that consumer knows where the Kafka is running and it can connect to the cluster so we will have the same set of properties here also now in this case because we have slightly different configurations sometimes for producer and consumer so we have two different classes for producer we use producer config and for Consumer we have something called consumer config but some properties are actually common so for example for the bootstrap we have the same property which is bootstrap do servers and here we will pass Local Host 9092 where the Kafka is running okay then we also need to pass the key D serializer and value Ser der serializer in this case so in this case it needs der serializer not serializer consumer config do e and you see D serializer class conf config okay and we will pass the value here because the key is integer so we need to use the same deserializer let me copy the class from simple producer this one and the only difference is this is going to be D serializer and it must be small like this then the third property is how the value will be deserialized so for that we need value deserializer and this is going to be the custom der serializer that we created so we will pass the name here like this okay now in case of consumers we also need to pass something called a group ID we will cover the concept of group ID and consumer groups probably in the next video so I will explain the concept there only for now let's pass the required config which is consumer config do group ID config and I can pass a random value here so I can say group hyphen a for example and then I will create the object of Kafka consumer by passing the set of properties that we created like this okay so once we have the Kafka consumer the next thing is basically to consume the events from the topic so what generally happens you write a long running consumer that will keep pulling basically the topic and in case there is a new event it will basically read the event so that's what we are going to do in this demo in future we will see variations of different types of consumers but for this one we will have a pooling consumer and since this is a long running so we will have a long running clop something like this while true okay and before we can consume the records we need to make an additional call which is subscribe so a consumer must subscribe a set of topics before it can read the events from those topics okay so this accepts a collection and because in this case we have a single event so we will pass a single item here which is user hyphen signups like this now what we can do here we have the Kafka consumer now so we can call the pool method and you can pass the duration so let's keep it small I can say of mes let's say uh window of 100 milliseconds so every 100 milliseconds it will pull the Kafka topic and in case there is a new event it will simply do something about it okay now when we pull the Kafka topic we get consumer records and we will pass the key and value so that we don't have to type cast explicitly the same key and value like this because it does not know when it is pulling the topic it may read multiple records so it returns a collection and then and once we have the collection of consumer records we can iterate on this collection and in each iteration we will get a particular consumer record again we need to pass the key and value type like this and record on records now this consumer record is actually the event that we are consuming and how do we get the event from this consumer record we not only get the exact event but we also get some additional metadata with the help of this consumer record so in the demo what we will do once we have the consumer record let's say we can print additional details along with the event itself like this record dot we can have let's say partition what is the partition from which we consume the event okay then we can also get the key because we are passing the key so that is simple we will get a valid key and to access the actual event we will call the method record. value so we have the simple consumer ready we have the simple producer ready and as I mentioned ear we are not dealing with exceptions we are not dealing with errors we are not closing the producers and consumers which we should do and which we will do in a practical project in a real project but for this one we are omitting that part okay so now that we have everything ready let's verify that the Kafka is running so we can check in the logs that the Kafka is indeed running all right let's start the consumer first because it will start pulling the topic and it will simply get blocked at this point of time which we want before we run the producer okay so let's run the consumer first and see if everything is working fine so we can see that the consumer is up it's not doing anything because it is waiting for an event to be available on the partition it is pulling and getting blocked at this line all right let's start the producer now so we see welcome to the portal let's enter the user ID we will say 101 name Demi user and it says that the event is sent now if we check the consumer we see the consumer consumed an event okay we see the partition which is zero because we have a single partition and partitions are zero based index so we see the partition number as zero then we get the key which is 101 and the value is the event itself now because we don't have the two string method in the event class so we don't see the exact details but the important thing is we are generating the events and we are consuming the events now let's try to send another event so we will see type c to continue okay enter another user ID let's enter a new user ID Dy user hyphen 1 and we see another event was sent let's check the consumer so we see a new event on the logs so in this video we learn how to develop a simple producer and a simple consumer and we also worked with Kafka topics and uh so in this video we'll learned how to develop simple CFA producers and consumers we learned how to use producer records consumer records how to pull a topic so on and so forth so I will see you in the next video thanks for watching