Build a Video Stream Microservice with Kafka & REST API in Java

To build a video stream processing microservice with a REST API in Java, using Kafka and Zookeeper, we can design an architecture where the REST API serves as an entry point to manage video streams. This microservice will:

Capture video frames and send them to Kafka.
Process video frames through Kafka topics.
Provide an API for starting and stopping streams, monitoring, and retrieving processed frames.

We’ll use Spring Boot for the REST API, Apache Kafka for message management, and OpenCV for video processing. Here’s a step-by-step guide.

Architecture Overview

REST API (Spring Boot) - A controller that handles API requests for starting/stopping video streams.
Kafka Cluster with Zookeeper - Manages video stream topics.
Video Producer - Captures video frames and sends them to Kafka.
Video Processor - Consumes, processes frames, and sends results to an output topic.
Consumer - Retrieves and displays or stores processed frames.

Here’s a simple example to demonstrate how Apache Kafka, Zookeeper, and Kafka Streams work together. This example will cover setting up Kafka with Zookeeper, creating a Kafka topic, and using Kafka Streams to process data in that topic.

Prerequisites

Apache Kafka and Zookeeper installed.
Java (preferably Java 8+).
A Kafka client library, such as kafka-clients and kafka-streams in your Java project.

Steps to Set Up and Run

Step 1: Start Zookeeper and Kafka

Start Zookeeper, then Kafka, as Kafka depends on Zookeeper.

bash

# Start Zookeeper

zookeeper-server-start.sh config/zookeeper.properties

# Start Kafka

kafka-server-start.sh config/server.properties

Step 2: Create Kafka Topics

Create topics for raw video frames and processed frames.

bash

kafka-topics.sh --create --topic raw-video-frames --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

kafka-topics.sh --create --topic processed-video-frames --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

Step 3: Set Up a Spring Boot Project

Dependencies (Maven pom.xml)

Include Kafka, Spring Boot, OpenCV, and other necessary dependencies:

xml

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-web</artifactId>

</dependency>

<groupId>org.springframework.kafka</groupId>

<artifactId>spring-kafka</artifactId>

</dependency>

<groupId>org.apache.kafka</groupId>

<artifactId>kafka-clients</artifactId>

</dependency>

<groupId>org.bytedeco</groupId>

<artifactId>opencv</artifactId>

</dependency>

</dependencies>

Step 4: Define Video Capture, Processing, and Producer Services

VideoProducerService: Captures video frames and sends them to Kafka.
VideoProcessorService: Consumes frames from raw-video-frames, applies processing (e.g., grayscale conversion), and publishes to processed-video-frames.
VideoController: Spring REST controller for managing the video stream.

Example Code

VideoProducerService

This service captures frames from a video source and sends them to the Kafka topic raw-video-frames.

java

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.ProducerRecord;

import org.opencv.core.Core;

import org.opencv.core.Mat;

import org.opencv.videoio.VideoCapture;

import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

import java.util.Properties;

@Service

public class VideoProducerService {

static { System.loadLibrary(Core.NATIVE_LIBRARY_NAME); }

private KafkaProducer<String, byte[]> producer;

private VideoCapture videoCapture;

@PostConstruct

public void init() {

Properties props = new Properties();

props.put("bootstrap.servers", "localhost:9092");

props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

producer = new KafkaProducer<>(props);

videoCapture = new VideoCapture(0); // Webcam, or replace with video path

}

public void startVideoStream() {

new Thread(() -> {

Mat frame = new Mat();

while (videoCapture.isOpened()) {

if (videoCapture.read(frame)) {

byte[] frameData = new byte[(int) (frame.total() * frame.channels())];

frame.get(0, 0, frameData);

producer.send(new ProducerRecord<>("raw-video-frames", frameData));

try { Thread.sleep(55); } catch (InterruptedException e) { e.printStackTrace(); }

}

}).start();

}

public void stopVideoStream() {

videoCapture.release();

}

VideoProcessorService

Consumes raw frames from Kafka, processes them (e.g., converts them to grayscale), and publishes them to processed-video-frames.

java

import org.apache.kafka.clients.consumer.ConsumerRecord;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.kafka.clients.producer.KafkaProducer;

import org.apache.kafka.clients.producer.ProducerRecord;

import org.opencv.core.CvType;

import org.opencv.core.Mat;

import org.opencv.core.Size;

import org.opencv.imgproc.Imgproc;

import org.springframework.stereotype.Service;

import java.util.Collections;

import java.util.Properties;

@Service

public class VideoProcessorService {

public void startProcessing() {

Properties consumerProps = new Properties();

consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "video-processor");

consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");

KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProps);

consumer.subscribe(Collections.singletonList("raw-video-frames"));

Properties producerProps = new Properties();

producerProps.put("bootstrap.servers", "localhost:9092");

producerProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

producerProps.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

KafkaProducer<String, byte[]> producer = new KafkaProducer<>(producerProps);

while (true) {

for (ConsumerRecord<String, byte[]> rec : consumer.poll(100)) {

Mat frame = new Mat(new Size(640, 480), CvType.CV_8UC3);

frame.put(0, 0, rec.value());

// Apply grayscale conversion

Mat grayFrame = new Mat();

Imgproc.cvtColor(frame, grayFrame, Imgproc.COLOR_BGR2GRAY);

byte[] processedFrameData = new byte[(int) (grayFrame.total() * grayFrame.channels())];

grayFrame.get(0, 0, processedFrameData);

producer.send(new ProducerRecord<>("processed-video-frames", processedFrameData));

}

VideoController

The REST controller allows you to start and stop the video stream through HTTP requests.

java

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.web.bind.annotation.*;

@RestController

@RequestMapping("/api/video")

public class VideoController {

@Autowired

private VideoProducerService producerService;

@Autowired

private VideoProcessorService processorService;

@PostMapping("/start")

public String startVideoStream() {

producerService.startVideoStream();

processorService.startProcessing();

return "Video stream started!";

}

@PostMapping("/stop")

public String stopVideoStream() {

producerService.stopVideoStream();

return "Video stream stopped!";

}

Step 5: Run the Spring Boot Application

Start your Spring Boot application, which will expose the REST API at http://localhost:8080/api/video.

Ø Start Stream: Send a POST request to http://localhost:8080/api/video/start to begin streaming and processing.

Ø Stop Stream: Send a POST request to http://localhost:8080/api/video/stop to stop streaming.

Step 6: Video Display or Storage Consumer

Finally, create a consumer application to fetch and display processed frames from processed-video-frames. You could integrate this with your frontend for real-time video display or save frames to storage.

Additional Considerations

Ø Concurrency: Run multiple instances of Kafka consumers for parallel processing.

Ø Error Handling: Add logging and error handling for production readiness.

Ø Data Format: For high performance, consider using Avro or Protobuf for video frame serialization.

This setup provides a REST API to control a video stream processing pipeline, making it suitable for real-time video applications with Kafka and Spring Boot in Java.