Spring Boot 4 Microservices Project: Home Energy Tracker - Ingestion Service & Kafka (Part 5)

This chapter covers the architectural design, implementation, and verification of the Ingestion Service within the Home Energy Tracker ecosystem. We explore how to ingest high-frequency telemetry from distributed IoT devices, stream those events reliably into Apache Kafka running in KRaft mode, manage local containerized infrastructure using Docker Compose, and build multi-threaded simulation components to stress-test our data pipeline.

1. Why Do We Need an Ingestion Service?

Decoupling Data Producers from Consumers: IoT devices and data producers must not block or depend on downstream database performance or heavy aggregations in usage-tracking microservices.
Handling Traffic Spikes: Thousands of smart meters and energy monitors emit telemetry continuously. An ingestion tier buffers these requests to protect downstream persistence layers from thrashing.
High Throughput Streaming: A dedicated gateway service cleanly accepts HTTP traffic, validates payloads, and hands off the data to an event backbone for asynchronous consumption.

CODE

IoT Smart Meters / Devices
   |
   | HTTP POST /api/v1/ingestion
   v
Ingestion Service (Spring Boot 4)
   |
   | publishes event
   v
Apache Kafka Event Stream

2. Core Mental Model: The Post Office Analogy

Real-Life Analogy: Imagine a busy postal sorting facility. Individual citizens (IoT devices) drop off thousands of mail envelopes at the front desk all day long. The receptionist (Ingestion Service) does not drive across town to deliver every single letter immediately. Instead, the receptionist drops all letters into large, secure canvas mailbags (Kafka Topics). Delivery drivers (Downstream Usage Services) pick up these mailbags whenever they have capacity, ensuring zero lost letters even if delivery trucks experience delays.

3. Creating the Ingestion Service Module & Dependencies

To bootstrap the new microservice module, use Spring Initializr targeting Spring Boot 4, Java, and Maven. Name the module ingestion-service and select the following fundamental starters:

Spring Web: Exposes HTTP REST endpoints.
Lombok: Reduces boilerplate code for constructors, builders, and utility methods.
Spring for Apache Kafka: Enables producer and consumer integration with Kafka brokers.

Essential `pom.xml` Additions for Spring Boot 4

Due to increased modularization in Spring Boot 4, certain foundational libraries like Jackson JSR-310 support for Java Instant objects must be declared explicitly:

XML

<dependencies>
    <!-- Spring Web, Lombok, Kafka Starters -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    
    <!-- Spring Boot 4 Modular Jackson & Date/Time Support -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>com.fasterxml.jackson.datatype</groupId>
        <artifactId>jackson-datatype-jsr310</artifactId>
    </dependency>
</dependencies>

4. Designing Immutable DTOs with Java Records

Data Transfer Objects (DTOs) act as pure data carriers. Because they contain no business logic and require immutability once instantiated, they are prime candidates for Java Records (standardized in Java 16).

JAVA

package com.lejourney.dto;

import java.time.Instant;

public record EnergyUsageDto(
    Long deviceId,
    double energyConsumed,
    Instant timestamp
) {}

Note: Entities managed by JPA cannot be records because records are immutable and lack standard no-args constructors and mutators required by ORM frameworks. DTOs and event payloads, however, benefit immensely from record syntax.

5. Implementing the Ingestion REST Controller

The REST controller exposes a single endpoint dedicated to high-volume telemetry ingestion. It returns an HTTP 201 Created status upon successfully queuing the incoming data.

JAVA

package com.lejourney.controller;

import com.lejourney.dto.EnergyUsageDto;
import com.lejourney.service.IngestionService;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/v1/ingestion")
public class IngestionController {

    private final IngestionService ingestionService;

    public IngestionController(IngestionService ingestionService) {
        this.ingestionService = ingestionService;
    }

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public void ingestData(@RequestBody EnergyUsageDto usageDto) {
        ingestionService.ingestEnergyUsage(usageDto);
    }
}

6. Running Apache Kafka in KRaft Mode via Docker Compose

Modern Kafka deployments eliminate Apache ZooKeeper by using KRaft (Kafka Raft Metadata Mode). Below is the configuration for running Kafka and Kafka UI inside Docker Compose.

YAML

services:
  kafka:
    image: apache/kafka:4.0.1
    container_name: kafka
    ports:
      - "9092:9092"
      - "9094:9094"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'INTERNAL://kafka:9092,EXTERNAL://localhost:9094'
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:9093'
      KAFKA_LISTENERS: 'INTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:9094'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'INTERNAL'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      CLUSTER_ID: 'Mk3OEYBSD3T24uJwvxWPDg=='
    volumes:
      - kafka_data:/var/lib/kafka/data

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    ports:
      - "8070:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local-kraft-cluster
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
    depends_on:
      - kafka

volumes:
  kafka_data:

Launching the Infrastructure

Open your terminal and run:

BASH

docker compose up -d

You can access the Kafka UI dashboard at http://localhost:8070 to monitor topics, consumer groups, and live message flows.

7. Connecting Spring Boot to Kafka & Publishing Events

To publish messages onto a Kafka topic, we define an event record that matches our DTO structure, inject a KafkaTemplate, and configure application properties.

Defining the Kafka Event Record

JAVA

package com.lejourney.kafkaevent;

import java.time.Instant;
import lombok.Builder;

@Builder
public record EnergyUsageEvent(
    Long deviceId,
    double energyConsumed,
    Instant timestamp
) {}

Application Properties Configuration (`application.properties`)

PROPERTIES

server.port=8082

# Kafka Producer Configuration
spring.kafka.bootstrap-servers=localhost:9094
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

Ingestion Service Implementation

JAVA

package com.lejourney.service;

import com.lejourney.dto.EnergyUsageDto;
import com.lejourney.kafkaevent.EnergyUsageEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class IngestionService {

    private static final String TOPIC_NAME = "energy-usage";
    private final KafkaTemplate<String, EnergyUsageEvent> kafkaTemplate;

    public IngestionService(KafkaTemplate<String, EnergyUsageEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void ingestEnergyUsage(EnergyUsageDto dto) {
        EnergyUsageEvent event = EnergyUsageEvent.builder()
                .deviceId(dto.deviceId())
                .energyConsumed(dto.energyConsumed())
                .timestamp(dto.timestamp())
                .build();

        // No key is passed as strict event ordering is not required for independent meter readings
        kafkaTemplate.send(TOPIC_NAME, event);
        log.info("Published energy usage event for device ID: {}", dto.deviceId());
    }
}

8. Simulating IoT Load: Sequential vs. Parallel Data Generators

To verify that our ingestion pipeline and Kafka cluster handle heavy burst loads without failure, we build simulation components implementing Spring Boot's CommandLineRunner and @Scheduled tasks.

A. Continuous Single-Threaded Simulator

This component fires sequential requests at configured intervals.

JAVA

package com.lejourney.simulation;

import com.lejourney.dto.EnergyUsageDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.time.Instant;
import java.util.Random;

@Slf4j
@Component
public class ContinuousDataSimulator implements CommandLineRunner {

    private final RestTemplate restTemplate = new RestTemplate();
    private final Random random = new Random();

    @Value("${simulation.endpoint}")
    private String ingestionEndpoint;

    @Value("${simulation.requests-per-interval}")
    private int requestsPerInterval;

    @Override
    public void run(String... args) {
        log.info("Continuous Data Simulator initialized.");
    }

    // @Scheduled(fixedRateString = "${simulation.interval-ms}")
    public void sendMockData() {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);

        for (int i = 0; i < requestsPerInterval; i++) {
            try {
                EnergyUsageDto dto = new EnergyUsageDto(
                        (long) random.nextInt(6) + 1,
                        random.nextDouble() * 200.0,
                        Instant.now()
                );

                HttpEntity<EnergyUsageDto> request = new HttpEntity<>(dto, headers);
                restTemplate.postForEntity(ingestionEndpoint, request, Void.class);
            } catch (Exception e) {
                log.error("Failed to send mock telemetry data", e);
            }
        }
        log.info("Sent batch of {} mock telemetry records.", requestsPerInterval);
    }
}

B. Parallel Multi-Threaded Simulator

To simulate realistic, concurrent device connections from multiple IoT gateways, we utilize an ExecutorService thread pool.

JAVA

package com.lejourney.simulation;

import com.lejourney.dto.EnergyUsageDto;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.time.Instant;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Component
public class ParallelDataSimulator implements CommandLineRunner {

    private final RestTemplate restTemplate = new RestTemplate();
    private final Random random = new Random();
    private ExecutorService executorService;

    @Value("${simulation.endpoint}")
    private String ingestionEndpoint;

    @Value("${simulation.requests-per-interval}")
    private int requestsPerInterval;

    @Value("${simulation.parallel-threads}")
    private int parallelThreads;

    @Override
    public void run(String... args) {
        this.executorService = Executors.newFixedThreadPool(parallelThreads);
        log.info("Parallel Data Simulator started with {} threads.", parallelThreads);
    }

    @Scheduled(fixedRateString = "${simulation.interval-ms}")
    public void sendMockDataParallel() {
        int batchSize = requestsPerInterval / parallelThreads;
        int remainder = requestsPerInterval % parallelThreads;

        for (int i = 0; i < parallelThreads; i++) {
            final int threadIndex = i;
            final int requestsForThread = batchSize + (threadIndex < remainder ? 1 : 0);

            executorService.submit(() -> {
                HttpHeaders headers = new HttpHeaders();
                headers.setContentType(MediaType.APPLICATION_JSON);

                for (int j = 0; j < requestsForThread; j++) {
                    try {
                        EnergyUsageDto dto = new EnergyUsageDto(
                                (long) random.nextInt(6) + 1,
                                random.nextDouble() * 150.0,
                                Instant.now()
                        );

                        HttpEntity<EnergyUsageDto> request = new HttpEntity<>(dto, headers);
                        restTemplate.postForEntity(ingestionEndpoint, request, Void.class);
                    } catch (Exception e) {
                        log.error("Error in parallel simulation thread", e);
                    }
                }
            });
        }
        log.info("Dispatched {} total requests across {} parallel threads.", requestsPerInterval, parallelThreads);
    }

    @PreDestroy
    public void shutdown() {
        if (executorService != null) {
            executorService.shutdown();
            log.info("Parallel Data Simulator executor service shut down.");
        }
    }
}

9. Enabling Scheduling & Simulator Properties

Remember to enable scheduling application-wide in your main Spring Boot class:

JAVA

package com.lejourney;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@EnableScheduling
@SpringBootApplication
public class IngestionServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(IngestionServiceApplication.class, args);
    }
}

Add the corresponding simulation properties to application.properties:

PROPERTIES

# Simulation Configuration
simulation.endpoint=http://localhost:8082/api/v1/ingestion
simulation.requests-per-interval=1000
simulation.parallel-threads=10
simulation.interval-ms=5000

10. Summary and Key Takeaways

Decoupled Architecture: The Ingestion Service acts as a high-speed gateway accepting IoT metrics via REST and pushing them asynchronously to Apache Kafka.
Java Records for DTOs: Immutable data carriers without business logic or JPA lifecycle requirements are cleanly and concisely modeled using Java records.
ZooKeeper-less Kafka (KRaft): Modern Kafka clusters run lightweight and efficiently in KRaft mode using internal quorum controllers.
Automatic Topic Provisioning: Setting KAFKA_AUTO_CREATE_TOPICS_ENABLE=true allows Spring Boot applications to automatically create topics upon first publication without manual pre-provisioning.
Kafka Key Selection Strategy: Omitting message keys distributes high-volume, unordered telemetry across partitions round-robin style, maximizing throughput.
Concurrent Load Simulation: Implementing CommandLineRunner, ExecutorService, and @Scheduled annotations allows engineers to simulate realistic multi-threaded IoT traffic bursts locally.
Modular Spring Boot 4 Dependencies: Explicitly adding starter JSON and Jackson JSR-310 libraries ensures correct serialization of Java 8 date/time types like Instant.

Spring Boot 4 Microservices Project - Home Energy Tracker - Ingestion Service & Kafka (Part 5)

Spring Boot 4 Microservices Project: Home Energy Tracker - Ingestion Service & Kafka (Part 5)

1. Why Do We Need an Ingestion Service?

2. Core Mental Model: The Post Office Analogy

3. Creating the Ingestion Service Module & Dependencies

Essential pom.xml Additions for Spring Boot 4

4. Designing Immutable DTOs with Java Records

5. Implementing the Ingestion REST Controller

6. Running Apache Kafka in KRaft Mode via Docker Compose

Launching the Infrastructure

7. Connecting Spring Boot to Kafka & Publishing Events

Defining the Kafka Event Record

Application Properties Configuration (application.properties)

Ingestion Service Implementation

8. Simulating IoT Load: Sequential vs. Parallel Data Generators

A. Continuous Single-Threaded Simulator

B. Parallel Multi-Threaded Simulator

9. Enabling Scheduling & Simulator Properties

10. Summary and Key Takeaways

Essential `pom.xml` Additions for Spring Boot 4

Application Properties Configuration (`application.properties`)