Spring Kafka Producer(Part 2): API and Producer Architecture

Goal: This is in continuation to part 1, we will be integrating Kafka to the API, to publish the notification payload to kafka broker.

https://medium.com/@prashantprakash97/spring-kafka-producer-part-1-api-and-producer-architecture-6f0f036c9a64

Prerequisites: I am starting this blog, where I left off last time. You can get the code for that from github:

https://github.com/Geek8080/notification-service/tree/3-2-notification-producer-API

Adding dependencies

We need only 1 dependency to get spring kafka working with our application. Go to pom.xml of producer module and add the following:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>${spring.kafka.version}</version>
</dependency>

Add the property for version into parent pom:

<spring.kafka.version>3.1.2</spring.kafka.version>

Configure connection to Kafka Brokers

I hope you have your kafka broker running. If not, go through this once:
https://medium.com/@prashantprakash97/spring-kafka-tutorial-setting-up-project-d979ea0ae26a

Go to offset explorer, and connect to this broker by entering the zookeeper details.

In my case, it is running on 192.168.1.12:2181, here’s the config page:

Connect to it, and you are good to go for this step.

Get Prashantprakash’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Next, we will create the configuration class, which will configure properties for the producer. It will look like below:

package io.Geek8080.notification.producer.config;

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Geek8080 on 02-03-2024.
 */

@Configuration
public class KafkaConfigurations {

    @Value("${spring.kafka.producer.broker}")
    public String kafkaBroker;

    @Bean
    public ProducerFactory<Integer, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaBroker);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // See https://kafka.apache.org/documentation/#producerconfigs for more properties
        return props;
    }

    @Bean
    public KafkaTemplate<Integer, String> kafkaTemplate() {
        return new KafkaTemplate<Integer, String>(producerFactory());
    }

}

This sets up the basic producer, which will connect to the broker defined by property “spring.kafka.producer.broker”. Go to application.properties and add the broker like this:

spring.kafka.producer.broker=192.168.1.12:9092

Now let’s write some code to push the data to Kafka topic, “email-notification”. We already have a controller, just go to that, autowire the kafka template, and send the data. And, yes, it is as simple as it sounds, here’s the code:

package io.Geek8080.notification.producer;

import com.google.gson.Gson;
import io.Geek8080.notification.dao.EmailNotificationRequest;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

/**
 * Created by Geek8080 on 17-02-2024.
 */

@RestController
@Log4j2
public class NotificationController {

    @Autowired
    private KafkaTemplate<Integer, String> kafkaTemplate;

    private static final String TOPIC = "email-notification";


    @PostMapping("sendEmail")
    public ResponseEntity<?> sendEmail(@RequestBody EmailNotificationRequest emailNotification) {
       log.info("Received a request to send email at: {}", emailNotification);

       kafkaTemplate.send(TOPIC, 1, new Gson().toJson(emailNotification));

       return ResponseEntity.ok("SUCCESS");
    }
}

Run your code, and open postman to hit this API:

If you check the logs, you will see just one log, which we have added:

2024-03-31 13:30:36,034 [http-nio-8080-exec-4] INFO  io.Geek8080.notification.producer.NotificationController.sendEmail():29 {} - Received a request to send email at: EmailNotificationRequest(emailId=usersame@gmail.com, message=hello Sam!)

Go to offset explorer, select the topic, and partition. Go to data tab and click on run button. You will see the data pushed:

Confused about the value? The thing is, by default, offset shows the hash. To get the complete data in plaintext, go to properties, and change the key and value to string, as follows:

Click on update button after setting the content type properties.

Now go back to data, and click on the run button again:

Finally, we can see the data we pushed.

We are done with the producer part. Simple, right? In next part we will start working on the consumer.

Code: https://github.com/Geek8080/notification-service/tree/3-3-notification-producer-API