Hands On: Receiving Messages with KafkaListener

Hi, this is Victor Gamov with Confluent. And in this lesson, I will show you how you can use Kafka-Listener to receive new messages from Kafka topic in Confluent Cloud. Let's get to it. In the previous lesson, you learnt how you can send a new messages into Kafka topic, and we use Confluent Cloud UI to consume those messages. And this is the time where you will learn how you can do this programmatically using Spring Boot and Spring Kafka. Let's create a new class called Consumer. In this class we will have some methods called public, void, consume. And in this methods, we need to do something with this data. So we can say it's gonna be string quote. And in this case, I will do KafkaListener annotation here. I need to specify what's the topic. In our case, it will be hobbit. You can subscribe to multiple topics as you know and group ID is important in the property that we need to specify. Spring-boot-Kafka Listener. Now, in order to do something with this quote, we will just do... And let's call it... We need to make this as a component and we're good to go. Also, we need to tell default deserializer what kind of data is coming into our application. So in this case, we can do kafka.consumer.value-deserializer. And in this case, we're gonna be using, org.apache.kafka.common.serialization.SringDeserializer. And so we not forget about the key deserializer as well. So we go in kafka.consumer.key-deserializer. And we need specified org.apache.kafka.common.serialization/IntangerDeserializer. Since we have a data in format and integer for key and the string for our value, I need to specify those deserializers here. In the future lessons I'll show you how we can use a more complex types and how we can use schemas with this application and how we can configure your application to use all this nice things. Now, my consumer is ready to roll. I need to go and restart my application. Takes some time to compile this. Location is up and running. And now we start seeing all this new messages arriving in this console. So let's take a look what do we see in Confluent Cloud. In the Confluent Cloud, we can go to this left-side menu and find the consumers. Sometimes it will take some time to refresh, so it's always good to refresh it. And now see the same group that I specified in my code over here saying that it is gonna be spring-boot-Kafka Listener. I see this consumer group is consuming all partitions, all six partitions in my topic, and I can see offsets, I can see consumer lag here, all these nice things. And also, if I go to this hobbit topic, I will be able to see a different dependencies or different applications that connected to this topic. In this particular case, that's my consumer application. Custom app. And that's my producer application. Also custom app. So essentially in my application configuration, my Spring Boot application configuration, I will be able to change this. By default, producer one is default client ID. You can change this in Spring Boot application. Like here, and you can go and say... Producer.client-id I can say spring-boot-producer. Also, if you're interested in your application, if you're interested to getting also information about the keys, you can do that as well. And let me show you how you can do that. So let me stop this application real quick. I'm getting back to my code. And instead of right now, instead of string, instead of value, I actually can't inject here and consumer record. And in this case, my consumer record will be key as an integer, and my value as a string. And I can say, record, and I can say this dot value, and something like plus with key, record.key. So now if I run this application once again, we should be able to get access to a record key. As we can see here, all this new messages start arriving with this updated information. My consumer is again, up and running, and we will be able to see what this consumer does. Now you know how you can consume messages from Confluent Cloud using Spring Boot and Kafka Listener.