Site icon Vinsguru

Kafka Stream With Spring Boot

vIns

Overview:

In this tutorial, I would like to show you how to do real time data processing by using Kafka Stream With Spring Boot.

Stream Processing:

In the good old days, we used to collect data, store in a database and do nightly processing on the data. It is called batch processing!

In this Microservices era, we get continuous / never ending stream of data. Sometimes delaying this data processing might have a severe impact in our business. For example, Let’s consider an application like Netflix / YouTube. Based on the movie/videos we surf, these applications show immediate recommendations. It provides much better user experience and helps with the business. Similarly when we get all the credit card transactions, a Bank might want to check if there is any fraudulent activity and block the card immediately if it is found! Credit card provider would not want to delay this as part of nightly processing.

Stream processing is a real time continuous data processing. Lets see how we can achieve a simple real time stream processing using Kafka Stream With Spring Boot.

Prerequisite:

A basic knowledge on Kafka is required. Read the below articles if you are new to this topic.

Sample Application:

To demo this real time stream processing, Lets consider a simple application which contains 3 microservices.

Producer, Processor and Consumer are 3 different applications connected via 2 different Kafka topics as shown below.

Kafka Set up:

Take a look at this article Kafka – Local Infrastructure Setup Using Docker Compose, set up a Kafka cluster. Once done, create 2 topics.

Spring Boot – Project Set up:

Java Functional Interface:

Spring Cloud Functions simplifies these application development by using below Java functional interfaces.

Application Type Java Functional Interface
Kafka Producer Supplier
Kafka Consumer Consumer
Kafka Processor Function

Kafka Stream Producer:

Working on Kafka Stream with Spring Boot is very easy! Spring Boot does all the heavy lifting with its auto configuration. I create a simple bean which will produce a number every second.

@Configuration
public class KafkaProducer {

    /*
    *   produce a number from 1, every second
    *   Supplier<T> makes this as kafka producer of T
    * */

    @Bean
    public Supplier<Flux<Long>> numberProducer(){
        return () -> Flux.range(1, 1000)
                        .map(i -> (long) i)
                        .delayElements(Duration.ofSeconds(1));
    };

}

Now important question is where would the data be written into? If you remember, we had created a topic for this – numbers. We configure that via application.yaml as shown below.

spring.cloud.stream:
  function:
    definition: numberProducer
  bindings:
    numberProducer-out-0:
      destination: numbers
      producer:
        use-native-encoding: true
  kafka:
    bindings:
      numberProducer-out-0:
        producer:
          configuration:
            value:
              serializer: org.apache.kafka.common.serialization.LongSerializer
    binder:
      brokers:
        - localhost:9091
        - localhost:9092

Kafka Stream Consumer:

As you had seen above, Spring Boot does all the heavy lifting. This is what I have to do to consume the data.

@Configuration
public class KafkaConsumer {

    /*
    *   consume the numbers received via kafka topic
    *   Consumer<T> makes this as kafka consumer of T
    * */

    @Bean
    public Consumer<KStream<String, Long>> squaredNumberConsumer(){
        return stream -> stream.foreach((key, value) -> System.out.println("Square Number Consumed : " + value));
    };

}
spring.cloud.stream:
  function:
    definition: squaredNumberConsumer
  bindings:
    squaredNumberConsumer-in-0:
      destination: squaredNumbers
  kafka:
    binder:
      brokers:
        - localhost:9091
        - localhost:9092

Kafka Stream Processor:

Processor is both Producer and Consumer. It consumes the data from 1 topic and produces data for another topic.

In our case, we have to do the following

Lets create the processor by using the corresponding Functional Interface in Java which is Function<T, R>. 

@Configuration
public class KafkaProcessor {

    /*
    *   process the numbers received via kafka topic
    *   Function<T, R> makes this as kafka stream processor
    *   T is input type
    *   R is output type
    *
    * */

    @Bean
    public Function<KStream<String, Long>, KStream<String, Long>> evenNumberSquareProcessor(){
        return kStream ->  kStream
                            .filter((k, v) -> v % 2 == 0)
                            .peek((k, v) -> System.out.println("Squaring Even : " + v))
                            .mapValues(v -> v * v);
    };

}
spring.cloud.stream:
  function:
    definition: evenNumberSquareProcessor
  bindings:
    evenNumberSquareProcessor-in-0:
      destination: numbers
    evenNumberSquareProcessor-out-0:
      destination: squaredNumbers
  kafka:
    binder:
      brokers:
        - localhost:9091
        - localhost:9092

Kafka Stream Processing:

Now at this point, everything seems to be ready. Start the applications. Once the apps are connected to the Kafka brokers, we can see the console outputs as shown below.

Squaring Even : 2
Squaring Even : 4
Squaring Even : 6
Squaring Even : 8
Squaring Even : 10
Squaring Even : 12
Squaring Even : 14
Square Number Consumed : 4
Square Number Consumed : 16
Square Number Consumed : 36
Square Number Consumed : 64
Square Number Consumed : 100
Square Number Consumed : 144
Square Number Consumed : 196

Bonus: Kafka + Spring Boot – Event Driven:

When we have multiple microservices with different data sources, data consistency among the microservices is a big challenge. You can take a look at this article how the problem is solved using Kafka for Spring Boot Microservices – here.

Summary:

We were able to successfully demonstrate real time data processing by using Kafka Stream with Spring Boot.

Learn more about Kafka + SpringBoot.

Check out other Streaming solutions – Redis Stream With Spring Boot – Real Time Data Processing

The source code is available here.

Happy learning 🙂

Share This:

Exit mobile version