Home Energy Tracker: Usage Service & InfluxDB Architecture (Part 6)

1. Architectural Role of the Usage Service

The Usage Service is the analytical core and central data processing hub of the Home Energy Tracker microservices architecture. It acts simultaneously as a Kafka Consumer, a Time-Series Database Writer, an Inter-Service Client, and a Kafka Producer.

Structure & Concept Flow

Ingestion Service -> Kafka Topic (energy-usage)

[ Usage Service ]

+---------------+---------------+

| |

v v

InfluxDB (2.7) User/Device Services

| |

+---------------+---------------+

Kafka Topic (energy-alerts) -> Alerting Service

Core Responsibilities

Kafka Consumer: Ingests high-volume energy consumption events (energy-usage) streamed from the Ingestion Service.
Time-Series Storage: Persists raw sensor metrics into InfluxDB for efficient range queries and time-series aggregation.
Microservices Orchestration: Queries the Device Service and User Service synchronously via HTTP (RestTemplate) to resolve device ownership and user alerting preferences.
Kafka Producer: Publishes aggregated threshold-breach events (energy-alerts) back to Kafka for downstream consumption by the Alerting Service.

2. Real-Life Analogy: The Smart Utility Smart-Meter Aggregator

Think of the Usage Service as a central power grid billing and safety bureau.

Millions of smart meters (simulated by the Ingestion Service) continuously dump raw meter readings into a massive post box (Kafka Topic).
An army of clerks reads these slips and files them into an ultra-fast filing cabinet optimized strictly for chronological data (InfluxDB).
Once every hour, a compliance auditor gathers the files, checks who owns which meter by calling the property registry (Device Service), reviews each owner's custom electricity budget contract (User Service), and if anyone went over budget, instantly mails an alarm notice to the dispatch queue (Kafka Alert Topic).

3. Provisioning InfluxDB in Docker Compose

To handle high-throughput time-series data without relational database bottlenecks, the project incorporates InfluxDB 2.7.

`docker-compose.yml` Configuration

YAML

version: '3.8'

services:
  influxdb:
    image: influxdb:2.7
    container_name: influxdb
    ports:
      - "8072:8086"
    environment:
      - DOCKER_INFLUXDB_INIT_MODE=setup
      - DOCKER_INFLUXDB_INIT_USERNAME=admin
      - DOCKER_INFLUXDB_INIT_PASSWORD=admin123
      - DOCKER_INFLUXDB_INIT_ORG=lejourney
      - DOCKER_INFLUXDB_INIT_BUCKET=usage_bucket
      - DOCKER_INFLUXDB_INIT_RETENTION=1w
      - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-token
    volumes:
      - influxdb-data:/var/lib/influxdb2

volumes:
  influxdb-data:

Note: The 1w retention policy ensures that raw operational metrics automatically expire after one week, preventing uncontrolled disk bloating. Long-term cold storage archiving can be implemented later in cloud object stores.

4. Setting Up the Usage Service Maven Module & Dependencies

The Usage Service is initialized as a standard Maven multi-module project running on Java 21 and Spring Boot 4.

`pom.xml` Dependencies

XML

<dependencies>
    <!-- Spring Web for REST clients and server -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!-- Spring for Apache Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>

    <!-- Lombok Boilerplate Reduction -->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>

    <!-- Jackson DataType for Java 8 / Instant support -->
    <dependency>
        <groupId>com.fasterxml.jackson.datatype</groupId>
        <artifactId>com.fasterxml.jackson.datatype:jackson-datatype-jsr310</artifactId>
    </dependency>

    <!-- InfluxDB Java Client -->
    <dependency>
        <groupId>com.influxdb</groupId>
        <artifactId>influxdb-client-java</artifactId>
        <version>6.12</version>
    </dependency>
</dependencies>

5. Consuming Kafka Events & Persisting to InfluxDB

`application.properties` (Consumer & Influx Settings)

PROPERTIES

server.port=8083

# Kafka Consumer Configuration
spring.kafka.bootstrap-servers=localhost:1994
spring.kafka.consumer.group-id=usage-service-group
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.properties.spring.json.trusted.packages=*

# InfluxDB Configuration
influx.url=http://localhost:8072
influx.token=my-token
influx.org=lejourney
influx.bucket=usage_bucket

InfluxDB Client Configuration Bean

JAVA

package com.lejourney.config;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.InfluxDBClientFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class InfluxDbConfig {

    @Value("${influx.url}")
    private String influxUrl;

    @Value("${influx.token}")
    private String influxToken;

    @Value("${influx.org}")
    private String influxOrg;

    @Bean
    public InfluxDBClient influxDBClient() {
        return InfluxDBClientFactory.create(influxUrl, influxToken.toCharArray(), influxOrg);
    }
}

Kafka Listener & Persistence Logic

JAVA

package com.lejourney.service;

import com.influxdb.client.InfluxDBClient;
import com.influxdb.client.WriteApiBlocking;
import com.influxdb.client.write.Point;
import com.influxdb.client.domain.WritePrecision;
import com.lejourney.kafka.event.EnergyUsageEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
@Slf4j
@RequiredArgsConstructor
public class UsageService {

    private final InfluxDBClient influxDBClient;

    @Value("${influx.bucket}")
    private String influxBucket;

    @Value("${influx.org}")
    private String influxOrg;

    @KafkaListener(topics = "energy-usage", groupId = "usage-service-group")
    public void consumeEnergyUsage(EnergyUsageEvent event) {
        // Build InfluxDB Data Point
        Point point = Point.measurement("energy-usage")
                .addTag("device_id", String.valueOf(event.getDeviceId()))
                .addField("energy_consumed", event.getEnergyConsumed())
                .time(event.getTimestamp(), WritePrecision.MS);

        // Write synchronously to InfluxDB bucket
        WriteApiBlocking writeApi = influxDBClient.getWriteApiBlocking();
        writeApi.writePoint(influxBucket, influxOrg, point);
    }
}

6. Scheduled Aggregation & Flux Query Execution

To detect threshold violations, a scheduled cron task executes periodically (set to every 10 seconds for rapid testing; production systems would typically run hourly or daily).

Enabling Scheduling in Spring Boot

JAVA

package com.lejourney;

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

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

Scheduled Aggregation Task & Flux Query

JAVA

package com.lejourney.service;

import com.influxdb.query.FluxRecord;
import com.influxdb.query.FluxTable;
import com.lejourney.client.DeviceClient;
import com.lejourney.client.UserClient;
import com.lejourney.dto.DeviceDto;
import com.lejourney.dto.UserDto;
import com.lejourney.kafka.event.AlertingEvent;
import com.lejourney.model.DeviceEnergy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
@RequiredArgsConstructor
public class UsageService {

    private final InfluxDBClient influxDBClient;
    private final DeviceClient deviceClient;
    private final UserClient userClient;
    private final KafkaTemplate<String, AlertingEvent> kafkaTemplate;

    @Value("${influx.bucket}")
    private String influxBucket;

    @Value("${influx.org}")
    private String influxOrg;

    @Scheduled(cron = "*/10 * * * * *")
    public void aggregateDeviceEnergyUsage() {
        Instant now = Instant.now();
        Instant oneHourAgo = now.minusSeconds(3600);

        // Construct InfluxDB Flux Query
        String fluxQuery = String.format(
            "from(bucket: \"%s\") " +
            "  |> range(start: %s, stop: %s) " +
            "  |> filter(fn: (r) => r[\"_measurement\"] == \"energy-usage\") " +
            "  |> filter(fn: (r) => r[\"_field\"] == \"energy_consumed\") " +
            "  |> group(columns: [\"device_id\"]) " +
            "  |> sum()",
            influxBucket, oneHourAgo.toString(), now.toString()
        );

        List<FluxTable> tables = influxDBClient.getQueryApi().query(fluxQuery, influxOrg);
        List<DeviceEnergy> deviceEnergies = new ArrayList<>();

        for (FluxTable table : tables) {
            for (FluxRecord record : table.getRecords()) {
                String deviceIdStr = (String) record.getValueByKey("device_id");
                Object valueObj = record.getValue();
                double energyConsumed = (valueObj instanceof Number) ? ((Number) valueObj).doubleValue() : 0.0;

                deviceEnergies.add(DeviceEnergy.builder()
                        .deviceId(Long.parseLong(deviceIdStr))
                        .energyConsumed(energyConsumed)
                        .build());
            }
        }
        
        // Enrichment and alerting processing follows...
    }
}

7. Inter-Service Communication: Resolving Devices & Users via REST

Once raw aggregated consumption numbers are retrieved from InfluxDB, the Usage Service determines which human owner is accountable by making synchronous HTTP calls to the Device Service and User Service.

Device Client & DTO

JAVA

package com.lejourney.client;

import com.lejourney.dto.DeviceDto;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

@Component
@RequiredArgsConstructor
public class DeviceClient {

    private final RestTemplate restTemplate = new RestTemplate();

    @Value("${device.service.url}")
    private String deviceServiceUrl;

    public DeviceDto getDeviceById(Long deviceId) {
        String url = UriComponentsBuilder.fromUriString(deviceServiceUrl)
                .path("/{id}")
                .buildAndExpand(deviceId)
                .toUriString();

        try {
            ResponseEntity<DeviceDto> response = restTemplate.getForEntity(url, DeviceDto.class);
            return response.getBody();
        } catch (Exception e) {
            return null; // Graceful failure handling
        }
    }
}

JAVA

package com.lejourney.dto;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class DeviceEnergy {
    private Long deviceId;
    private double energyConsumed;
    private Long userId;
}

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class DeviceDto {
    private Long id;
    private String name;
    private String type;
    private String location;
    private Long userId;
}

User Client & DTO

JAVA

package com.lejourney.client;

import com.lejourney.dto.UserDto;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

@Component
@RequiredArgsConstructor
public class UserClient {

    private final RestTemplate restTemplate = new RestTemplate();

    @Value("${user.service.url}")
    private String userServiceUrl;

    public UserDto getUserById(Long userId) {
        String url = userServiceUrl + "/" + userId;
        try {
            ResponseEntity<UserDto> response = restTemplate.getForEntity(url, UserDto.class);
            return response.getBody();
        } catch (Exception e) {
            return null;
        }
    }
}

JAVA

package com.lejourney.dto;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class UserDto {
    private Long id;
    private String name;
    private String surname;
    private String email;
    private String address;
    private boolean alerting;
    private Double energyAlertingThreshold;
}

8. Threshold Evaluation & Producing Alerting Events

After mapping devices to users and summing their total power usage across all owned hardware meters, the Usage Service checks whether the user's defined consumption threshold has been breached.

JAVA

// Inside UsageService.java aggregation method:

// 1. Enrich devices with userId from Device Service
for (DeviceEnergy de : deviceEnergies) {
    try {
        DeviceDto deviceDto = deviceClient.getDeviceById(de.getDeviceId());
        if (deviceDto != null && deviceDto.getUserId() != null) {
            de.setUserId(deviceDto.getUserId());
        }
    } catch (Exception e) {
        log.warn("Failed to fetch device ID {}: {}", de.getDeviceId(), e.getMessage());
    }
}

// 2. Remove devices without valid user associations
deviceEnergies.removeIf(de -> de.getUserId() == null);

// 3. Group by User ID
Map<Long, List<DeviceEnergy>> userDeviceMap = deviceEnergies.stream()
        .collect(Collectors.groupingBy(DeviceEnergy::getUserId));

// 4. Check against User Thresholds and Publish Kafka Alert
userDeviceMap.forEach((userId, devices) -> {
    try {
        UserDto user = userClient.getUserById(userId);
        if (user != null && user.isAlerting() && user.getEnergyAlertingThreshold() != null) {
            double totalConsumption = devices.stream().mapToDouble(DeviceEnergy.getEnergyConsumed).sum();

            if (totalConsumption > user.getEnergyAlertingThreshold()) {
                AlertingEvent alertEvent = AlertingEvent.builder()
                        .userId(userId)
                        .message("Energy consumption threshold exceeded!")
                        .threshold(user.getEnergyAlertingThreshold())
                        .energyConsumed(totalConsumption)
                        .email(user.getEmail())
                        .build();

                kafkaTemplate.send("energy-alerts", alertEvent);
                log.info("Alert published for User ID {} - Total Consumption: {}", userId, totalConsumption);
            }
        }
    } catch (Exception e) {
        log.error("Failed to process user alerting for ID {}: {}", userId, e.getMessage());
    }
});

Alerting Event Record

JAVA

package com.lejourney.kafka.event;

import lombok.Builder;

@Builder
public record AlertingEvent(
    Long userId,
    String message,
    Double threshold,
    Double energyConsumed,
    String email
) {}

9. Comparison: InfluxDB vs. Relational Databases for Time-Series Metrics

Feature / Dimension	Relational Database (MySQL / PostgreSQL)	Time-Series Database (InfluxDB 2.7)
Data Ingestion Rate	Moderate; b-tree index degradation under high write loads.	Extreme; optimized append-only columnar storage.
Storage Footprint	Large; high index overhead per row.	Highly compressed via delta and run-length encoding.
Data Retention	Requires custom cron jobs and `DELETE` queries.	Native built-in automated bucket retention policies.
Time-Range Queries	Requires complex index scans and range predicates.	Extremely fast native windowing and aggregation (`sum()`, `mean()`).

10. Industry Reality & Debugging Edge Cases

When running distributed microservices processing high-velocity telemetry, several failure modes commonly surface in production environments:

Orphaned Sensor Readings (Device Not Found): If an ingestion event arrives for a device ID that was never registered in the Device Service, REST lookups return 404 or null. Solution: Wrap inter-service calls in robust try-catch blocks and filter out unresolvable entities rather than letting exceptions crash the entire scheduled aggregation batch.
Missing User Alert Configurations: Users may have active devices but have never configured an alert threshold or opted out of notifications (alerting = false). Always guard threshold comparisons with explicit null checks and boolean state validations.
Port Conflicts: Ensure each microservice runs on a distinct port (8080 for Ingestion, 8081 for Device, 8082 for User, and 8083 for Usage Service) and that Kafka bootstrap servers match the exact Docker port mapping (localhost:1994).

11. Summary and Key Takeaways

- Usage Service Architecture: Functions as a Kafka consumer for raw usage metrics, a time-series writer to InfluxDB, an HTTP orchestrator across Device and User services, and a Kafka producer for downstream alerts.
- InfluxDB 2.7 Integration: Deployed via Docker Compose with customized initialization parameters, admin tokens, bucket definitions, and a 1-week data retention policy.
- Kafka Consumer Configuration: Consumes energy-usage topics using Spring Kafka with JSON deserialization mapped to the EnergyUsageEvent record.
- Scheduled Aggregation: Utilizes Spring's @EnableScheduling and @Scheduled annotations to query InfluxDB at regular intervals using Flux query syntax.
- Distributed Orchestration: Employs RestTemplate and dedicated Feign-like client components to fetch device ownership and user alert thresholds synchronously.
- Fault Tolerance & Data Cleansing: Gracefully handles missing device records and inactive user alert states to ensure batch aggregations continue uninterrupted.
- Downstream Alert Publishing: Publishes serialized AlertingEvent payloads to the energy-alerts Kafka topic whenever a user's total hourly consumption exceeds their pre-configured threshold.

Spring Boot 4 Microservices Project - Home Energy Tracker - Usage Service & Influx DB (Part 6)

Home Energy Tracker: Usage Service & InfluxDB Architecture (Part 6)

1. Architectural Role of the Usage Service

Core Responsibilities

2. Real-Life Analogy: The Smart Utility Smart-Meter Aggregator

3. Provisioning InfluxDB in Docker Compose

docker-compose.yml Configuration

4. Setting Up the Usage Service Maven Module & Dependencies

pom.xml Dependencies

5. Consuming Kafka Events & Persisting to InfluxDB

application.properties (Consumer & Influx Settings)

InfluxDB Client Configuration Bean

Kafka Listener & Persistence Logic

6. Scheduled Aggregation & Flux Query Execution

Enabling Scheduling in Spring Boot

Scheduled Aggregation Task & Flux Query

7. Inter-Service Communication: Resolving Devices & Users via REST

Device Client & DTO

User Client & DTO

8. Threshold Evaluation & Producing Alerting Events

Alerting Event Record

9. Comparison: InfluxDB vs. Relational Databases for Time-Series Metrics

10. Industry Reality & Debugging Edge Cases

11. Summary and Key Takeaways

`docker-compose.yml` Configuration

`pom.xml` Dependencies

`application.properties` (Consumer & Influx Settings)