Project Setup and Kafka Module
Hey there. Today we're going to take a look at using KafkaJS in a Nest.js project to be able to connect to a Kafka server, produce messages, and consume messages. So I'm going to show you how to integrate this into an app and kind of my preferred approach to working with it.
In a new directory, we're going to, of course, start off a new project by using nest new and we'll use kafka-js. So in a new directory, I'll use the Nest CLI and create a new project, and I'll call this nestjs-kafka. So go ahead and let this finish scaffolding, and I'll use yarn for our package manager. So once that finishes, we can cd into the directory, and then I'm going to open up a code editor, and then we can also start the development server by running yarn start:dev.
So the first thing we're going to go ahead and do is add the KafkaJS package. So in a new terminal, I'm going to yarn add kafkajs. So now we have KafkaJS in our project. I'm also going to use the Nest CLI to generate a Kafka module. So we're going to call nest generate module kafka. So now we have KafkaJS and we have this new Kafka module. We're essentially going to set up two services that are applicable to work with: a producer service and a consumer service.
Building the Kafka Producer Service
So to start, let's start off with a producer service. So we'll create a producer service here. And this will be an injectable, and we'll export the producer service here.
Now, the first thing we want to do is actually get a handle to the Kafka server. And the way we do that is we can actually declare a private read-only variable here. We'll call it kafka, and we're going to set this equal to new Kafka, and this is from kafkajs, as you can see up above. And now we need to specify the brokers or the servers that our Kafka server is listening on. So when we start our server up locally, we're going to use localhost and it'll be the default port, which by default is 9092. So if you're running locally, this is the broker you should use.
So now we have a handle to the Kafka server, we actually want to actually get a handle to the producer we're going to make now too. So I'm going to declare another read-only private variable called producer of type Producer, and it's simply going to be this.kafka.producer(). So now we have the producer here, we need to actually connect to this producer. So we want to do that when the application starts up, and we can take advantage of the Nest.js application lifecycle hooks.
So we're going to make sure this service implements OnModuleInit. That way we can actually have a bootstrap method to hook into the onModuleInit here. And this can be an asynchronous method, which is very useful for our purposes because we need to call this.producer.connect(), which is an asynchronous method. So this line here will allow us to actually connect the producer to our server.
Now, we're going to add another async method to actually produce messages. We'll just call it produce, and it's going to take a record, and this record is of type ProducerRecord from KafkaJS. Okay, and then we can simply call await this.producer.send() and then pass in the record. That's really all we need to do.
We're just going to add one more lifecycle hook up here, actually OnApplicationShutdown, so that when the application shuts down, we can disconnect the producer. So we'll call onApplicationShutdown and we'll call await this.producer.disconnect(). So that when our Nest.js application stops, we disconnect the producer.
Make sure in our Kafka module we actually instantiate this as a provider, the producer service here, and we'll also add it to our exports array so we can use it in our app in a bit.
Building the Kafka Consumer Service
So now we have a producer service we can use to produce messages. Let's add a consumer service that allows us to consume messages. Okay, and this will be a little bit more involved, but not terribly difficult. So this will also be an injectable if you want to maintain state here, and we'll call this the Consumer Service.
So I'm going to go into the Producer service and just copy this line here where we get a handle to the Kafka server, and we can import Kafka again. Now we also now want a read-only variable and we want to get an array of consumers here, because in the producer's case, we only really need one producer to produce messages to our server, but in the consumer's case, we're going to have multiple consumers potentially that can subscribe to different topics. On the other hand, for the producer, the record itself here specifies the topic that we send the message to and the messages themselves. So in the consumer service, this will be an array of consumers because we want to actually track the amount of consumers we have in our application so that we can disconnect from them when the server shuts down. So this consumer will come from KafkaJS, and then we'll add an async consume method here.
So this consume method is going to take the topic, and this is of type ConsumerSubscribeTopic, okay, and the config is the ConsumerRunConfig. So if you want to take a closer look at what these look like, the topic itself is just the topic name, which is a string or regular expression, and the fromBeginning boolean, which instructs Kafka to read from the beginning of the topic when it starts to consume messages. And then the config itself is where we're going to specify the implementation of the method we want to run when a message is received. So we allow this to be defined in each of our implementations later on.
So now on the consume method, let's go ahead and actually create a consumer. We'll call this.kafka.consumer() and we need to specify the group ID here. So in this case, this could be scoped to the application we're currently using. So I'm just going to call this nestjs-kafka because that's our app we're currently using. So once you've created the consumer, we can call consumer.connect() as we have done with the producer, so that it actually connects the consumer to our server. And now we need to call consumer.subscribe() and we're going to subscribe to the topic that we've passed in here, so that it listens to the correct topic. And lastly, we're going to call await consumer.run() and this is where we pass in the config that our implementation will pass in, so that it will run certain code whenever a message is received.
So finally, we're going to actually push this consumer to our array of consumers. And the reason we're going to do that is because similarly to our producer service, we want to implement OnApplicationShutdown so that we can disconnect from these consumers when we shut down. So simply, we can iterate over the consumers. For const consumer of this.consumers, we're going to await consumer.disconnect(). And we don't need this signal parameter here.
So now we have a service that we can use in our code to easily consume messages and disconnect when our application shuts down. So as we've done before, we'll have to open up the Kafka module and let's add the consumer service here and also export it so that we can use it in our code.
Implementing the Producer and Consumer Logic
So now let's actually use these services we just created. In our App Controller, let's actually change the return type here, and we're going to go into the App Service, into our getHello method, and let's change this to an asynchronous method so we actually produce a message in this method here in our GET route. So we'll remove the return type as well.
And now in our App Service, we want to inject the Producer service. So let's do that. We'll inject the producer service here so we can make use of it, the Producer Service. And now in our getHello method, before we return, we can call await this.producerService.produce(). And now we have the producer record that we can provide. And as we've seen before, we'll have to provide a topic name. So in this case, I'll use a test topic and call it test, whatever we'd like. And then we have an array of messages that we would want to produce. So in this case, we're just going to send one message. And then we have the value here we can provide. I will provide 'Hello world'. So now we are correctly producing a message to the test topic.
So now that we're producing properly, we need a way to consume messages. So let's open up a new file here. We will call this the TestConsumer. And I like to call this or scope this by the name of the topic we have in this case. So in this case it's test, we'll have a test.consumer. So this will be an injectable class as always, so we can inject the Consumer Service. So export the TestConsumer. And importantly, we're going to have to implement the OnModuleInit here as we've done before. But firstly, we're going to need to inject the Consumer Service here. So let's do that first, the Consumer Service.
And now we can implement an asynchronous onModuleInit, and this is where we want to actually subscribe to the topic at hand using the Consumer Service, right when the application starts up. So let's call await this.consumerService.consume(). And now we can pass in the topic or the ConsumerSubscribeTopic, which in this case will have the topic name inside. So we have the topic name, and now we can provide the ConsumerRunConfig. And this is where we can provide the eachMessage property, which is the code we want to run when we receive a new message on this topic.
So in this case, it's going to be an asynchronous method. It doesn't have to be, but I'll make it async because typically you want to do async work in here. And then the object we get back is the topic, has the topic inside, it has the partition, and it has the message itself. So then we can go ahead and inside of this method, we'll simply console.log an object, and we'll include the value, which is message.value.toString(), the topic itself as topic.toString(), and then the partition is partition.toString().
So now this test consumer, when our application starts up, it's going to subscribe to our test topic and execute this code whenever a message is received. So finally, don't forget to add this as a provider in our providers array here in the AppModule.
Starting the Local Kafka Server
So lastly, we need to make sure we start up our Kafka server locally. If you don't have Kafka already installed, I'll include the link here in the description. A pretty simple installation. We just need to get the latest Kafka release, unzip it, and then cd into the directory where we have a couple of scripts that we can run to start the server.
So if you already have Kafka installed, let's go ahead and start up the server. You're just going to cd into that Kafka folder that you have installed. And then once you're inside of this directory, we're going to run this command to start the Zookeeper server using Zookeeper properties. Make sure you start up the Zookeeper server first. And then you're going to open up another terminal and go to that same directory to start the actual Kafka server. So I'm going to cd right back into the Kafka directory that I have installed, paste this next command which starts the Kafka server using server properties. So let's go ahead and let the server start up.
Now lastly, let's go ahead and stop our NestJS server and let's restart it all together. So I'm going to run yarn start:dev to restart the server and you should see logs here from KafkaJS stating that the consumer has joined the group. So our group ID that we specified earlier, nestjs-kafka, is here, and we can see it's using the RoundRobinAssigner. So that our consumer has correctly joined the application.
Final Demonstration and Recap
So let's test this out. If we open up Postman, we should be able to make a GET request on localhost:3000, which is where our server is listening. And remember, we set up our producer to produce a message when this code is executed. So if we send off a few requests here, we can see the consumer service is correctly logging out the message from the producer. We have the 'Hello World' value here with the topic and the partition.
So using these services, we can very easily set up any amount of consumers and producers we want anywhere in our code by importing this Kafka module here and following this simple example. I hope you have learned something. So thanks so much and be sure to like, comment, and subscribe if you enjoyed the video. And I'll see you in the next one. Thanks.
