Building Scalable Applications with Kafka and Reactive Programming Mahi Mullapudi

Introduction

In today's digital world, the ability to process data in real time and scale applications is critical. Kafka, a distributed event streaming platform, is a good fit for this, especially when combined with reactive programming. This article will show you how to build reactive applications using this toolkit.

Let's go over the basics

Before we get into the details, let's explore some key concepts.

Kafka: Apache Kafka is a distributed event stream processing platform used to build real-time data pipelines and streaming applications. Kafka processes huge amounts of data quickly. Kafka enables applications to produce and consume streams of records (events or messages).

Reactive programming: is an approach to software processing that emphasizes processing asynchronous data streams and propagating changes. This approach promotes responsiveness, reliability, and scalability of applications.

Scalability: This is the ability of a system to handle increasing workloads or the potential for the system to grow to accommodate that growth. In the context of an application, scalability is the ability to handle increasing numbers of users, data volumes, or new transactions without sacrificing performance.

Real-time streaming data processing: is a technology for processing data as it arrives, allowing an application to respond immediately to changes or events. Stream processing is usually contrasted with batch processing, in which data is accumulated and processed in batches at predetermined intervals.

Setup: Why Kafka and Reactive Programming?

Kafka and reactive programming complement each other well. Kafka provides a robust mechanism for processing data streams, while reactive programming allows you to process these streams efficiently and quickly. Together, these two technologies are useful for building scalable, real-time applications.

Key benefits:

1. High throughput: Kafka processes large volumes of data with minimal latency.
2. fault tolerance: Kafka's distributed architecture ensures data replication and system fault tolerance.
3. Scalability: Both Kafka and reactive programming are designed to scale horizontally, meaning you can add more and more nodes to handle increasing loads.
4. Asynchronous processing: Since reactive programming is asynchronous by nature, this paradigm guarantees non-blocking operations and improves responsiveness.

Setting up Kafka

To start programming scalable applications with Kafka, you first need to set up a Kafka cluster. A cluster consists of multiple Kafka brokers, each of which is responsible for processing data streams.

Step by step setup:

• Download and install Kafka:
• Let's move on download page Apache Kafka and take the latest version from there.
• Extract the downloaded files and go to the directory with Kafka.

• Launch Zookeeper:
• Kafka relies on Zookeeper, a distributed coordination service, to run. Let's start Zookeeper with the following command:
  sh bin/zookeeper-server-start.sh config/zookeeper.properties

• Launch the Kafka broker:
• Once Zookeeper is running, we start the Kafka broker:
  sh bin/kafka-server-start.sh config/server.properties

• Let's create a topic:
• Data in Kafka is organized into topics. Let's create a new topic like this:
  sh bin/kafka-topics.sh --create --topic my-topic --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

• We start producing and consuming messages:
• We produce messages in the topic:
  sh bin/kafka-console-producer.sh --topic my-topic --bootstrap-server localhost:9092
• Consuming messages from the topic:
  sh bin/kafka-console-consumer.sh --topic my-topic --bootstrap-server localhost:9092 --from-beginning

Introduction to Reactive Programming

Reactive programming concepts:

• Asynchrony: operations are non-blocking, that is, no operation must wait for another operation to complete.
• The driving role of events: the code reacts to events – that is, to user actions or to data changes.
• Observed flows: data is understood as flows that can be observed and responded to.

Popular libraries:

• Project Reactor: a library for reactive programming in Java, provides a powerful API for building asynchronous, event-driven applications.
• RxJava: another popular library for reactive programming in Java, inspired by the Reactive Extensions project.

Simple reactive example

Let's look at a simple example of reactive programming using Project Reactor.

Java

import reactor.core.publisher.Flux;

public class ReactiveExample {
    public static void main(String[] args) {
        Flux<String> dataStream = Flux.just("Hello", "Reactive", "World");
        dataStream
            .map(String::toUpperCase)
            .subscribe(System.out::println);
    }
}

Here Flux is a stream of data. The map function transforms each of the elements, and subscribe consumes the data, printing it to the console.

Integrating Kafka with Reactive Programming

To create a scalable application, we need to integrate Kafka with reactive programming tools. To do this, we will use a reactive Kafka client that will produce and consume messages.

Message production for Kafka

Using the reactor-kafka library, you can produce messages for Kafka in a reactive manner.

Dependencies:
Let's add the following dependencies to the pom.xml file (for Maven):

Java

<dependency>
    <groupId>io.projectreactor.kafka</groupId>
    <artifactId>reactor-kafka</artifactId>
    <version>1.3.5</version>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.7.0</version>
</dependency>

Kafka Reactive Producer:

Java

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import reactor.core.publisher.Flux;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;

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

public class ReactiveKafkaProducer {
    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        SenderOptions<String, String> senderOptions = SenderOptions.create(props);
        KafkaSender<String, String> sender = KafkaSender.create(senderOptions);

        Flux<SenderRecord<String, String, String>> outboundFlux = Flux.range(1, 10)
            .map(i -> SenderRecord.create(new ProducerRecord<>("my-topic", "key" + i, "value" + i), "correlationId" + i));

        sender.send(outboundFlux)
            .doOnError(e -> System.err.println("Send failed: " + e))
            .doOnNext(r -> System.out.println("Message sent: " + r.correlationMetadata()))
            .subscribe();
    }
}

In this example, we set up a Kafka producer using reactive programming. It sends ten messages to the topic my-topic, logging whether each operation succeeded or failed.

Consuming messages from Kafka

Similarly, you can reactively consume messages from Kafka.

Reactive Kafka Consumer:

Java

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;

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

public class ReactiveKafkaConsumer {
    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "reactive-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(props)
            .subscription(Collections.singleton("my-topic"));

        KafkaReceiver<String, String> receiver = KafkaReceiver.create(receiverOptions);

        receiver.receive()
            .doOnNext(record -> System.out.println("Received message: " + record.value()))
            .subscribe();
    }
}

In this example, the consumer subscribes to my-topic and prints all received messages to the console.

Building a Scalable Application: An Example Analysis

To illustrate how Kafka can be used in conjunction with reactive programming, let's look at a practical example: let's write a system that allows real-time tracking of an entire fleet of trucks. The system must process data on the movement of the vehicles and show where each of them is at any given moment.

System components

1. Truck sensors: Sensors installed in vehicles that send location and status updates to the system.
2. Kafka Broker: collects information from sensors and sends it in streaming mode.
3. Reactive Microservices: process and analyze data.
4. Data Visualization Application: shows the user data updated in real time.

Step by step implementation:

1. Truck sensors: Let's simulate sensors sending information to Kafka.

Java

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import reactor.core.publisher.Flux;
import reactor.kafka.sender.KafkaSender;
import reactor.kafka.sender.SenderOptions;
import reactor.kafka.sender.SenderRecord;

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

public class TruckSensorSimulator {
    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        props.put(Producer

Config.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        SenderOptions<String, String> senderOptions = SenderOptions.create(props);
        KafkaSender<String, String> sender = KafkaSender.create(senderOptions);

        Random random = new Random();
        Flux<SenderRecord<String, String, String>> sensorDataFlux = Flux.interval(Duration.ofSeconds(1))
            .map(tick -> {
                String truckId = "truck-" + random.nextInt(10);
                String location = "loc-" + random.nextInt(100);
                String status = "status-" + random.nextInt(3);
                String value = String.format("%s,%s,%s", truckId, location, status);
                return SenderRecord.create(new ProducerRecord<>("truck-data", truckId, value), "correlationId" + tick);
            });

        sender.send(sensorDataFlux)
            .doOnError(e -> System.err.println("Send failed: " + e))
            .doOnNext(r -> System.out.println("Message sent: " + r.correlationMetadata()))
            .subscribe();
    }
}

2.

Kafka Broker

: let's prepare and start Kafka as described above.

3. Reactive Microservices: we will process data from Kafka and analyze it in real time.

Java

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import reactor.kafka.receiver.KafkaReceiver;
import reactor.kafka.receiver.ReceiverOptions;

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

public class TruckDataProcessor {
    public static void main(String[] args) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "truck-data-processor-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

        ReceiverOptions<String, String> receiverOptions = ReceiverOptions.create(props)
            .subscription(Collections.singleton("truck-data"));

        KafkaReceiver<String, String> receiver = KafkaReceiver.create(receiverOptions);

        receiver.receive()
            .doOnNext(record -> {
                String[] data = record.value().split(",");
                String truckId = data[0];
                String location = data[1];
                String status = data[2];
                // Process and analyze the data (e.g., updating a database or triggering alerts)
                System.out.println("Processed data for truck: " + truckId + ", location: " + location + ", status: " + status);
            })
            .subscribe();
    }
}

4.

Data Visualization Application

: we show processed data to the user in real time.

To build such an application, you can use a web framework that supports reactive programming, such as Spring Boot with WebFlux. The application will subscribe to a WebSocket endpoint to receive real-time updates.

Setting up Spring Boot WebFlux:

Dependencies:

Let's add the following dependencies to the pom.xml file (for Maven):

Java

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.7.0</version>
</dependency>

WebFlux Configuration:

Java

import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.config.EnableWebFlux;
import org.springframework.web.reactive.config.WebFluxConfigurer;

@Configuration
@EnableWebFlux
public class WebFluxConfig implements WebFluxConfigurer {
    // WebFlux configuration can be added here if needed
}

WebSocket configuration:

Java

import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new TruckDataWebSocketHandler(), "/truck-data").setAllowedOrigins("*");
    }
}

WebSocket Handler:

Java

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
import reactor.core.publisher.Mono;

public class TruckDataWebSocketHandler extends WebSocketHandlerAdapter implements WebSocketHandler {

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        // Simulate sending real-time updates to the client
        return session.send(
            session.receive()
                .map(msg -> session.textMessage("Received: " + msg.getPayloadAsText()))
                .doOnError(e -> System.err.println("WebSocket error: " + e))
        );
    }
}

Reactive Kafka Consumer Service:

Java

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;

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

@Service
public class TruckDataService {

    private final Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

    @KafkaListener(topics = "truck-data", groupId = "truck-data-processor-group")
    public void listen(String message) {
        sink.tryEmitNext(message);
    }

    public Flux<String> getTruckDataStream() {
        return sink.asFlux();
    }

    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "truck-data-processor-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    public ConcurrentMessageListenerContainer<String, String> kafkaListenerContainer() {
        ContainerProperties containerProps = new ContainerProperties("truck-data");
        return new ConcurrentMessageListenerContainer<>(consumerFactory(), containerProps);
    }
}

REST controller:

Java

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

@RestController
public class TruckDataController {

    @Autowired
    private TruckDataService truckDataService;

    @GetMapping("/truck-data")
    public Flux<String> getTruckData() {
        return truckDataService.getTruckDataStream();
    }
}

With this setup, real-time vehicle movement data is sent to a WebSocket endpoint that a data visualization application can connect to, which is where we'll see the updates.

Conclusion

By combining Kafka and reactive programming, you can build highly scalable applications that can process and respond to real-time data streams. Kafka provides a robust mechanism for processing large volumes of data, while reactive programming ensures that the application remains responsive, resilient, and efficient.

This article covered how to get Kafka up and running, provided some basic information about reactive programming, and explained how to integrate the two to create a real-time monitoring system. If you're writing an application for monitoring, data processing, or other real-time tasks, Kafka and reactive programming are the tools you need to succeed.

By mastering these technologies, you can build applications that not only scale well as demand grows, but also provide real-time data and the responsiveness that users demand in today's dynamic digital environment.

P.S. Please note that we are currently running a sale on our website.