Threat Event Pipeline with Spring Boot

Build a SIEM pipeline that ingests security events via REST and streams through Kafka.

Introduction

30 Second Summary

Security teams monitor thousands of events per second across their networks. A single suspicious login attempt needs to be captured, classified by severity, and surfaced as an alert before an attacker can move deeper.

In this project, you will build a threat event ingestion pipeline that accepts security events via a REST API, streams them through Kafka for real-time processing, analyzes their severity with detection rules, and stores enriched alerts in PostgreSQL.

What You'll Build

Submit a suspicious login event and watch it flow through your pipeline in real time, get classified as HIGH severity by your detection rules, and appear in your alerts endpoint ready for a security analyst to act on.

By the end of this project, you'll have:

  • A REST API that ingests security events (failed logins, port scans, brute-force attempts), persists them to PostgreSQL, and streams them through Apache Kafka in real time.
  • A threat analysis engine that consumes events from Kafka, applies severity classification rules (LOW, MEDIUM, HIGH, CRITICAL), and escalates repeat offenders automatically.
  • A fully containerized local stack running with a single docker compose up command, complete with interactive Swagger UI docs, Spring Actuator health checks, and passing unit and integration tests.
  • Secret Mission: Extract the threat analyzer into a standalone second microservice that consumes from Kafka independently.

Are there any prerequisites?

You should be familiar with REST APIs, Docker, and basic database concepts. No prior Spring Boot or Kafka experience is required.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Development Environment

Before you can build a threat detection pipeline, you need a properly scaffolded Spring Boot project with the right dependencies.

In this step, you will generate a Spring Boot project from Spring Initializr, add extra dependencies for API documentation and testing, and spin up PostgreSQL and Kafka containers locally using Docker Compose.

In this step, get ready to:
  • Confirm JDK 17, Maven, and Docker Desktop are ready.
  • Generate the Spring Boot project from Spring Initializr.
  • Set up Docker Compose with PostgreSQL and Kafka.
Confirm your development tools

You should already have JDK 17+, Maven 3.6.3+, and Docker Desktop installed from the project prerequisites.

  • Open your terminal by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Terminal, and pressing Enter.
  • Confirm your Java version by running:
java -version
  • You should see version 17 or higher in the output.
  • Confirm your Maven version by running:
mvn -v
  • You should see version 3.6.3 or higher.
  • Confirm Docker Desktop is running by running:
docker info
  • You should see Docker server information without errors, confirming the daemon is running.

Seeing 'command not found' or connection errors?

If java -version fails, install JDK 17 with brew install openjdk@17 (macOS) or download from the Adoptium website (Windows).

If mvn -v fails, install Maven with brew install maven (macOS) or follow the official Apache Maven installation guide (Windows).

If docker info shows a connection error, open Docker Desktop and wait for it to fully start before retrying.

Help me troubleshoot my Java, Maven, or Docker installation.

Generate the Spring Boot project

Spring Initializr is a web tool that scaffolds a ready-to-run Spring Boot project with your chosen dependencies pre-configured.

  • Open start.spring.io in your browser.
  • Set Project to Maven.
  • Set Language to Java.
  • Set Spring Boot to 3.5.16.
  • Under Project Metadata, enter com.threatpipeline for Group.
  • Enter threat-ingest for Artifact.
  • Set Packaging to Jar and Java to 17.
  • Click ADD DEPENDENCIES and search for each of these, adding them one at a time:
  • Spring Web.
  • Spring Data JPA.
  • Spring for Apache Kafka.
  • Spring Boot Actuator.
  • PostgreSQL Driver.
  • Validation.
  • Click GENERATE to download the project as a zip file.
  • Extract the zip file to your Desktop.

What did Spring Initializr just create?

The downloaded zip contains a complete Maven project structure with a pom.xml (dependency manifest), a main application class, and the standard src/main/java and src/test/java directories.

Spring Boot auto-configures most settings based on which dependencies are present. You added the building blocks for a web API, database persistence, event streaming, health monitoring, and input validation.

  • Open your code editor by pressing Cmd+Space (macOS) or the Windows key (Windows), typing Cursor (or VS Code), and pressing Enter.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the threat-ingest folder on your Desktop, then click Open.
  • Confirm the project structure exists by opening the integrated terminal with Ctrl+` and running:
ls
  • You should see pom.xml, src, and mvnw in the output.
Add dependencies and create Docker Compose

Spring Initializr gives you the core dependencies, but this project also needs OpenAPI documentation, structured logging, and test infrastructure powered by Testcontainers.

  • In the Explorer sidebar, click pom.xml to open it.
  • Find the existing spring-kafka-test dependency block (near the bottom of the <dependencies> section).
  • Add the following dependencies directly below it:
        <dependency>
            <groupId>org.springdoc</groupId>
            <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
            <version>2.8.17</version>
        </dependency>
        <dependency>
            <groupId>net.logstash.logback</groupId>
            <artifactId>logstash-logback-encoder</artifactId>
            <version>8.0</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-testcontainers</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testcontainers</groupId>
            <artifactId>junit-jupiter</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testcontainers</groupId>
            <artifactId>postgresql</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.testcontainers</groupId>
            <artifactId>kafka</artifactId>
            <scope>test</scope>
        </dependency>

What do these dependencies do?

  • springdoc-openapi-starter-webmvc-ui auto-generates interactive API documentation (Swagger UI) from your controllers.
  • logstash-logback-encoder formats application logs as structured JSON for production observability.
  • The Testcontainers libraries spin up real PostgreSQL and Kafka containers during integration tests, so your tests run against the same infrastructure as production.
  • Now find the <plugins> section inside <build> (near the bottom of the file).
  • Add this plugin after the existing spring-boot-maven-plugin:
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <parameters>true</parameters>
                </configuration>
            </plugin>

Why add the compiler parameters flag?

Spring Boot 3.2+ relies on compiled parameter names for dependency injection and data binding. The <parameters>true</parameters> flag preserves method parameter names in the compiled bytecode, preventing hard-to-debug injection failures.

  • Save pom.xml.
  • Verify all dependencies download successfully by running:
mvn compile
  • You should see BUILD SUCCESS at the end of the output. Maven downloads all declared dependencies on the first run, so this may take a minute.

Seeing BUILD FAILURE?

Check that the <version>2.8.17</version> tag is present on the springdoc dependency. Without it, Maven cannot resolve the artifact.

Confirm your new dependencies are inside the <dependencies> block (not accidentally outside it). XML nesting errors cause silent parse failures.

Help me fix my Maven build failure after adding dependencies to pom.xml.

Now you need local infrastructure for your app to connect to. Docker Compose lets you define and run multi-container services from a single file.

  • Create a new file called docker-compose.yml in the root of your threat-ingest folder (next to pom.xml).
  • Add the PostgreSQL service definition by pasting this code:
services:
  postgres:
    image: postgres:16
    container_name: threat-postgres
    environment:
      POSTGRES_DB: threatdb
      POSTGRES_USER: threat
      POSTGRES_PASSWORD: threat123
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U threat -d threatdb"]
      interval: 5s
      timeout: 3s
      retries: 5

What does this service do?

  • image: postgres:16 pulls the official PostgreSQL 16 container image.
  • The environment variables create a database called threatdb with credentials your Spring Boot app will use to connect.
  • The healthcheck uses pg_isready to verify the database is accepting connections before reporting healthy.
  • Add the Kafka service and volumes section directly below the postgres service:

  kafka:
    image: apache/kafka:3.9.2
    container_name: threat-kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      CLUSTER_ID: MkU3OEVBNTcwNTJENDM2Qk
    volumes:
      - kafkadata:/var/lib/kafka/data
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 || exit 1"]
      interval: 10s
      timeout: 5s
      retries: 5
      start_period: 30s

volumes:
  pgdata:
  kafkadata:

What is KRaft mode?

Kafka traditionally required a separate ZooKeeper cluster for coordination. KRaft mode eliminates that dependency by running Kafka's metadata management internally. The KAFKA_PROCESS_ROLES: broker,controller line makes this single node handle both roles.

The replication factors are set to 1 because this is a single-node development setup. In production you would have multiple brokers with higher replication.

  • Save docker-compose.yml.
  • Start the containers in detached mode by running:
docker compose up -d
  • You should see Docker pulling the postgres:16 and apache/kafka:3.9.2 images, then creating and starting both containers.

Before you run the next command, what do you expect the health status of both containers to be?

  • Verify both containers are healthy by running:
docker compose ps
  • You should see both threat-postgres and threat-kafka with status healthy in the output. Kafka may take 30-40 seconds to reach healthy status due to start_period.

Containers not starting or showing unhealthy?

If a container shows as starting, wait 30 seconds and run docker compose ps again. Kafka needs time for KRaft initialization.

If port 5432 is already in use, you may have a local PostgreSQL running. Stop it first, or change the host port in docker-compose.yml to something like 5433:5432.

Check container logs with docker compose logs kafka or docker compose logs postgres to see detailed error messages.

Help me troubleshoot my Docker Compose containers for PostgreSQL and Kafka.

✔️ Awesome, I've got everything!

Great. Double check you have saved both pom.xml and docker-compose.yml.

ⓧ I'd like to double check the full code

Your pom.xml is over 30 lines, so scroll through and confirm each section is present in this order:

  • Parent block with spring-boot-starter-parent version 3.5.16.
  • GroupId com.threatpipeline and artifactId threat-ingest.
  • Java version property set to 17.
  • Dependencies: spring-boot-starter-web, spring-boot-starter-data-jpa, spring-kafka, spring-boot-starter-actuator, spring-boot-starter-validation, postgresql, springdoc-openapi-starter-webmvc-ui (2.8.17), logstash-logback-encoder (8.0), spring-boot-starter-test, spring-kafka-test, spring-boot-testcontainers, testcontainers junit-jupiter, testcontainers postgresql, testcontainers kafka.
  • Build plugins: spring-boot-maven-plugin and maven-compiler-plugin with <parameters>true</parameters>.

Your docker-compose.yml should contain these sections in order:

  • The postgres service with image postgres:16, environment variables, port 5432, volume, and healthcheck.
  • The kafka service with image apache/kafka:3.9.2, KRaft environment variables, port 9092, volume, and healthcheck.
  • The volumes block declaring pgdata and kafkadata.

Your development environment is fully set up. You have a scaffolded Spring Boot project and local infrastructure ready to accept connections. Next up, you will build the REST API that ingests threat events and persists them to PostgreSQL.

Build the Threat Event REST API and Persistence Layer

Your containers are running and your project is scaffolded. Now you need something for your pipeline to actually ingest.

The core of any SIEM platform is event ingestion. Security events (failed logins, port scans, brute-force attempts) need to land in a database the instant they arrive so nothing is lost before analysis. In this step, you will build a REST API that accepts raw security events, validates them, and persists them to PostgreSQL.

In this step, get ready to:
  • Create a ThreatEvent JPA entity with an EventType enum and a Spring Data repository.
  • Build a REST controller with POST and GET endpoints plus Jakarta Validation on the request DTO.
  • Configure application.yml to connect to your Docker Compose PostgreSQL and test the full round-trip.
Create the event model and repository

Every security event in your system needs a type classification. You will define the four event types your SIEM recognises as a Java enum, then build the JPA entity that maps each event to a PostgreSQL row.

  • In your project's src/main/java/com/threatpipeline/threatingest/ directory, create a new folder called model.
  • Inside the model folder, create a file called EventType.java with the following code:
package com.threatpipeline.threatingest.model;

public enum EventType {
    FAILED_LOGIN,
    PORT_SCAN,
    BRUTE_FORCE,
    MALWARE_DETECTED
}

What does this enum represent?

Each value maps to a category of security threat your pipeline can detect. FAILED_LOGIN tracks suspicious authentication attempts. PORT_SCAN catches network reconnaissance. BRUTE_FORCE flags repeated forced entry. MALWARE_DETECTED signals confirmed malicious software.

  • Save EventType.java.
  • Now create a file called ThreatEvent.java in the same model folder. Start with the entity class and its field declarations:
package com.threatpipeline.threatingest.model;

import jakarta.persistence.*;
import java.time.Instant;
import java.util.UUID;

@Entity
@Table(name = "threat_events")
public class ThreatEvent {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(nullable = false)
    private String sourceIp;

    @Enumerated(EnumType.STRING)
    @Column(nullable = false)
    private EventType eventType;

    @Column(nullable = false)
    private String targetSystem;

    @Column(nullable = false)
    private Instant timestamp;

    @Column(columnDefinition = "TEXT")
    private String rawPayload;

What do these annotations do?

@Entity tells Spring Data JPA this class maps to a database table. @Table names it threat_events. @Id marks the primary key, and @GeneratedValue(strategy = GenerationType.UUID) auto-generates a unique ID for every row.

@Enumerated(EnumType.STRING) stores the event type as readable text (like "BRUTE_FORCE") instead of a number. @Column(nullable = false) enforces that the field cannot be null at the database level.

  • Below the field declarations (still inside the class), add the constructor and accessor methods:

    public ThreatEvent() {}

    public UUID getId() { return id; }
    public void setId(UUID id) { this.id = id; }
    public String getSourceIp() { return sourceIp; }
    public void setSourceIp(String sourceIp) { this.sourceIp = sourceIp; }
    public EventType getEventType() { return eventType; }
    public void setEventType(EventType eventType) { this.eventType = eventType; }
    public String getTargetSystem() { return targetSystem; }
    public void setTargetSystem(String targetSystem) { this.targetSystem = targetSystem; }
    public Instant getTimestamp() { return timestamp; }
    public void setTimestamp(Instant timestamp) { this.timestamp = timestamp; }
    public String getRawPayload() { return rawPayload; }
    public void setRawPayload(String rawPayload) { this.rawPayload = rawPayload; }
}

Why do JPA entities need a no-arg constructor?

JPA uses reflection to create entity instances when loading data from the database. It needs the empty constructor to instantiate the object first, then populate the fields via setters. The getters and setters give Spring (and Jackson for JSON serialisation) access to each property.

  • Save ThreatEvent.java.
  • Confirm both model files exist by running:
ls src/main/java/com/threatpipeline/threatingest/model/

You should see EventType.java and ThreatEvent.java listed.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

Your ThreatEvent.java should combine the field declarations chunk and the accessor methods chunk into one file. Scroll up and confirm both pieces are present, starting with the package declaration and ending with the closing brace after setRawPayload.

  • Create a new folder called repository inside src/main/java/com/threatpipeline/threatingest/.
  • Inside the repository folder, create ThreatEventRepository.java with the following code:
package com.threatpipeline.threatingest.repository;

import com.threatpipeline.threatingest.model.EventType;
import com.threatpipeline.threatingest.model.ThreatEvent;
import org.springframework.data.jpa.repository.JpaRepository;
import java.util.List;
import java.util.UUID;

public interface ThreatEventRepository extends JpaRepository<ThreatEvent, UUID> {
    // Spring Data generates the SQL for this query from the method name
    List<ThreatEvent> findByEventType(EventType eventType);
}

What is Spring Data JPA doing here?

By extending JpaRepository<ThreatEvent, UUID>, you get CRUD methods (save, findAll, findById, delete) for free. Spring generates the implementation at runtime.

The custom method findByEventType is a query derivation. Spring parses the method name and generates a SQL WHERE clause that filters by the eventType column.

  • Save ThreatEventRepository.java.

Seeing red underlines in your IDE?

Your IDE might show errors because the project has not been compiled yet. This is expected at this point. The errors will resolve once all files are in place and Maven compiles the project.

If the import for JpaRepository does not resolve, confirm your pom.xml includes spring-boot-starter-data-jpa and reload Maven dependencies (click the refresh icon in your IDE's Maven panel).

Why is my repository interface not resolving?

Create the request DTO, service, and controller

The model layer stores data. Now you need the layers that receive HTTP requests, validate them, and orchestrate persistence. You will build a DTO (Data Transfer Object) for request validation, a service for business logic, and a controller that exposes the REST endpoints.

  • Create a new folder called dto inside src/main/java/com/threatpipeline/threatingest/.
  • Inside the dto folder, create CreateEventRequest.java with validation annotations:
package com.threatpipeline.threatingest.dto;

import com.threatpipeline.threatingest.model.EventType;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;

public class CreateEventRequest {

    @NotBlank(message = "sourceIp is required")
    private String sourceIp;

    @NotNull(message = "eventType is required")
    private EventType eventType;

    @NotBlank(message = "targetSystem is required")
    private String targetSystem;

    private String rawPayload;

    public String getSourceIp() { return sourceIp; }
    public void setSourceIp(String sourceIp) { this.sourceIp = sourceIp; }
    public EventType getEventType() { return eventType; }
    public void setEventType(EventType eventType) { this.eventType = eventType; }
    public String getTargetSystem() { return targetSystem; }
    public void setTargetSystem(String targetSystem) { this.targetSystem = targetSystem; }
    public String getRawPayload() { return rawPayload; }
    public void setRawPayload(String rawPayload) { this.rawPayload = rawPayload; }
}

What do the validation annotations do?

@NotBlank rejects null values AND empty strings. It ensures sourceIp and targetSystem always contain real data. @NotNull only checks for null (not empty), which is appropriate for the enum field since enums cannot be blank.

Spring Boot automatically returns a 400 Bad Request with error details when validation fails, so your API rejects garbage input before it hits the database.

  • Save CreateEventRequest.java.
  • Create a new folder called service inside src/main/java/com/threatpipeline/threatingest/.
  • Inside the service folder, create ThreatEventService.java to handle the persistence logic:
package com.threatpipeline.threatingest.service;

import com.threatpipeline.threatingest.dto.CreateEventRequest;
import com.threatpipeline.threatingest.model.EventType;
import com.threatpipeline.threatingest.model.ThreatEvent;
import com.threatpipeline.threatingest.repository.ThreatEventRepository;
import org.springframework.stereotype.Service;
import java.time.Instant;
import java.util.List;

@Service
public class ThreatEventService {

    private final ThreatEventRepository eventRepository;

    public ThreatEventService(ThreatEventRepository eventRepository) {
        this.eventRepository = eventRepository;
    }

    // Converts the request DTO into an entity and persists it
    public ThreatEvent createEvent(CreateEventRequest request) {
        ThreatEvent event = new ThreatEvent();
        event.setSourceIp(request.getSourceIp());
        event.setEventType(request.getEventType());
        event.setTargetSystem(request.getTargetSystem());
        event.setRawPayload(request.getRawPayload());
        event.setTimestamp(Instant.now());
        return eventRepository.save(event);
    }

What does this service do?

The service sits between the controller and the repository. It maps incoming request data onto a ThreatEvent entity, stamps it with the current time via Instant.now(), and delegates persistence to the repository's save() method. Spring returns the saved entity with its auto-generated UUID.

  • Below the createEvent method (still inside the class), add the query method and close the class:

    // Returns all events, or filters by type if specified
    public List<ThreatEvent> getEvents(EventType eventType) {
        if (eventType != null) {
            return eventRepository.findByEventType(eventType);
        }
        return eventRepository.findAll();
    }
}
  • Save ThreatEventService.java.
  • Confirm your project compiles with all the new files by running:
mvn compile -q

If everything is wired correctly, you will see BUILD SUCCESS. This confirms your model, repository, DTO, and service all compile together without errors.

Seeing BUILD FAILURE?

Check that every file is in the correct package folder. The package declaration at the top of each file must match its directory path exactly. For example, package com.threatpipeline.threatingest.model must live inside src/main/java/com/threatpipeline/threatingest/model/.

Also verify that the import statements in ThreatEventService.java reference the correct package paths for your DTO and model classes.

Help me fix my Maven compile error.

Now you need the controller that exposes the HTTP endpoints. This is what external clients (and your curl tests) will talk to.

  • Create a new folder called controller inside src/main/java/com/threatpipeline/threatingest/.
  • Inside the controller folder, create ThreatEventController.java with the class skeleton and POST endpoint:
package com.threatpipeline.threatingest.controller;

import com.threatpipeline.threatingest.dto.CreateEventRequest;
import com.threatpipeline.threatingest.model.EventType;
import com.threatpipeline.threatingest.model.ThreatEvent;
import com.threatpipeline.threatingest.service.ThreatEventService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.swagger.v3.oas.annotations.tags.Tag;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.web.bind.annotation.*;
import java.util.List;

@RestController
@RequestMapping("/api/events")
@Tag(name = "Threat Events", description = "Ingest and query raw security events")
public class ThreatEventController {

    private final ThreatEventService eventService;

    public ThreatEventController(ThreatEventService eventService) {
        this.eventService = eventService;
    }

What do these class-level annotations mean?

@RestController tells Spring this class handles HTTP requests and returns JSON. @RequestMapping("/api/events") sets the base URL path for all endpoints in this controller.

@Tag is an OpenAPI annotation that groups these endpoints under "Threat Events" in the auto-generated Swagger documentation. It does not affect runtime behaviour.

  • Below the constructor (still inside the class), add the two endpoint methods and close the class:

    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    @Operation(summary = "Ingest a security event", description = "Validates, persists, and publishes the event to Kafka")
    @ApiResponse(responseCode = "201", description = "Event created and published")
    public ThreatEvent createEvent(@Valid @RequestBody CreateEventRequest request) {
        return eventService.createEvent(request);
    }

    @GetMapping
    @Operation(summary = "Query threat events", description = "Returns all events, optionally filtered by type")
    public List<ThreatEvent> getEvents(@RequestParam(required = false) EventType eventType) {
        return eventService.getEvents(eventType);
    }
}

What do these endpoint annotations do?

@PostMapping maps HTTP POST requests to the createEvent method. @ResponseStatus(HttpStatus.CREATED) makes the response return 201 instead of the default 200. @Valid triggers the Jakarta Validation annotations on the DTO before the method body runs.

@GetMapping maps GET requests. The @RequestParam(required = false) makes the eventType query parameter optional. When omitted, the service returns all events.

  • Save ThreatEventController.java.

✔️ Awesome, I've got everything!

Great. Confirm you have saved both chunks (the class skeleton and the endpoint methods) in the same file.

ⓧ I'd like to double check the full code

Your ThreatEventController.java combines both chunks above into one file. Scroll up and confirm: the file starts with the package declaration and imports, includes the class-level annotations and constructor from the first chunk, then both endpoint methods from the second chunk, and ends with a closing brace.

Configure the application and test the endpoints

Your Java code is ready, but Spring Boot does not know where to find your PostgreSQL database yet. You need an application.yml configuration file that points the app at the Docker Compose containers you started in Step 1.

  • Create a file called application.yml inside src/main/resources/ (replace the existing application.properties if Spring Initializr generated one). Start with the Spring datasource and JPA configuration:
spring:
  application:
    name: threat-ingest
  datasource:
    url: jdbc:postgresql://localhost:5432/threatdb
    username: threat
    password: threat123
  jpa:
    hibernate:
      ddl-auto: create-drop
    show-sql: false
    properties:
      hibernate:
        format_sql: true
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
    consumer:
      group-id: threat-analyzer-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: com.threatpipeline.threatingest.model

What does this configuration do?

The datasource section points at your Docker Compose PostgreSQL container (port 5432, database threatdb, credentials threat/threat123). These match the environment variables in your docker-compose.yml.

The ddl-auto: create-drop setting tells Hibernate to auto-create tables on startup and drop them on shutdown. This is perfect for development but you would never use it in production. The Kafka section pre-configures the broker connection for a later step.

  • Below the spring section, add the management and documentation configuration:

management:
  endpoints:
    web:
      exposure:
        include: health,info,metrics
  endpoint:
    health:
      show-details: always

springdoc:
  api-docs:
    path: /v3/api-docs
  swagger-ui:
    path: /swagger-ui.html
  • Save application.yml.

Time to see if your pipeline's ingestion layer actually works. Before you run the app, what do you think will happen when Spring Boot starts? Consider that it needs to connect to PostgreSQL and create tables.

  • Make sure your Docker containers are still running by checking:
docker compose ps

You should see both threat-postgres and threat-kafka with status healthy.

  • Start the Spring Boot application by running:
mvn spring-boot:run

After a few seconds, you should see a log line indicating the application started successfully on port 8080. The app will keep running in this terminal.

Application fails to start?

If you see a connection refused error for port 5432, your PostgreSQL container may have stopped. Run docker compose up -d to restart it.

If you see a bean creation error mentioning KafkaProducerService, remove the kafkaProducerService reference from ThreatEventService for now. You will add Kafka integration in the next step.

Help me fix my Spring Boot startup error.

  • Open a second terminal (keep the app running in the first one).
  • Submit a test security event by running this curl command:
curl -X POST http://localhost:8080/api/events \
  -H "Content-Type: application/json" \
  -d '{
    "sourceIp": "192.168.1.100",
    "eventType": "FAILED_LOGIN",
    "targetSystem": "auth-server-01",
    "rawPayload": "{\"attempts\": 5}"
  }'

You should receive a JSON response with a 201 status containing your event data plus an auto-generated id (UUID) and a timestamp.

  • Now retrieve the persisted event by running:
curl http://localhost:8080/api/events

You should see a JSON array containing the event you just submitted, confirming it was persisted to PostgreSQL and retrieved successfully.

What just happened end-to-end?

Your curl POST hit the controller, which validated the request body (checking that sourceIp, eventType, and targetSystem are not blank/null). The service created a ThreatEvent entity with the current timestamp, and the repository persisted it to the threat_events table in PostgreSQL. The GET endpoint queried that same table and returned the stored data.

  • Test the validation by sending a bad request (missing required fields):
curl -X POST http://localhost:8080/api/events \
  -H "Content-Type: application/json" \
  -d '{"sourceIp": "", "targetSystem": ""}'

You should receive a 400 Bad Request response with validation error messages. This proves your Jakarta Validation annotations are working.

Getting a 404 instead of 201?

Confirm your controller class has @RestController and @RequestMapping("/api/events") annotations. Also check that the controller package (com.threatpipeline.threatingest.controller) is a sub-package of your main application class. Spring Boot component scanning only finds beans in the same package or sub-packages.

Why is my endpoint returning 404?

Your ingestion layer is live. Security events flow in via REST, get validated, and land safely in PostgreSQL. Next up, you will wire these persisted events into Kafka so downstream consumers can analyse them in real time.

Add Kafka Event Streaming

Your REST API persists events to PostgreSQL, which means nothing is lost. But persisting alone is not enough for a real-time SIEM.

Without an event stream, any downstream analyzer would need to poll the database for new records. That creates latency and tightly couples the analyzer to the ingestion layer. In this step, you'll add Apache Kafka as the streaming backbone so events flow to consumers in real time.

In this step, get ready to:
  • Create a Kafka producer service that publishes persisted events to a topic.
  • Create a Kafka consumer service that receives events from the topic.
  • Verify the full produce-consume loop in the application logs.
Create the Kafka producer service

Your application.yml already configures the Kafka bootstrap server, JSON serializer, and consumer group from the previous step. Now you need the service class that actually sends events to a Kafka topic.

  • Create a new file KafkaProducerService.java in src/main/java/com/threatpipeline/threatingest/service/ and paste the following code:
package com.threatpipeline.threatingest.service;

import com.threatpipeline.threatingest.model.ThreatEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    private static final Logger log = LoggerFactory.getLogger(KafkaProducerService.class);
    private static final String TOPIC = "threat-events";

    private final KafkaTemplate<String, ThreatEvent> kafkaTemplate;

    public KafkaProducerService(KafkaTemplate<String, ThreatEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishEvent(ThreatEvent event) {
        kafkaTemplate.send(TOPIC, event.getId().toString(), event);
        log.info("Published event to topic: {} with id: {}", TOPIC, event.getId());
    }
}

What does this code do?

  • KafkaTemplate<String, ThreatEvent> is Spring's abstraction for sending messages to Kafka. Spring auto-configures it using your application.yml serializer settings.
  • kafkaTemplate.send(TOPIC, key, value) publishes the event to the threat-events topic, using the event's UUID as the message key. Kafka uses the key to determine which partition receives the message.
  • The log.info line confirms each publish operation in the console, which you'll use to verify the pipeline works.

Why is ThreatEventService already wired?

If you open ThreatEventService.java, you'll see it already injects KafkaProducerService and calls kafkaProducerService.publishEvent(saved) after persisting every event. The class you just created satisfies that dependency.

  • Save the file.
Create the Kafka consumer service

The producer pushes events into Kafka. Now you need a consumer on the other side that pulls events out and processes them. For now, the consumer will just log each event it receives. In the next step, you'll replace that log with actual threat analysis logic.

  • Create a new file KafkaConsumerService.java in the same service/ package and paste the following code:
package com.threatpipeline.threatingest.service;

import com.threatpipeline.threatingest.model.ThreatEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerService.class);

    @KafkaListener(topics = "threat-events", groupId = "threat-analyzer-group")
    public void consumeEvent(ThreatEvent event) {
        log.info("Consumed event from Kafka: {}", event.getId());
        // Placeholder: analysis logic will be added in the next step
    }
}

What does this code do?

  • @KafkaListener(topics = "threat-events", groupId = "threat-analyzer-group") tells Spring to subscribe this method to the threat-events topic. Every message published by the producer triggers this method automatically.
  • The groupId identifies this consumer as part of the threat-analyzer-group. In a real system, multiple instances of the analyzer could share this group to load-balance event processing across them.
  • Spring deserializes the JSON message back into a ThreatEvent object using the trusted-packages config you set in application.yml.

Why produce and consume in the same app?

Both the producer and consumer live in the same Spring Boot application right now, but they communicate through Kafka rather than direct method calls. This simulates how separate microservices interact in production. The consumer's separate groupId means it operates independently from the producer.

In the Secret Mission, you'll extract this consumer into a standalone second microservice that runs on its own port.

  • Save the file.
Test the produce-consume loop

With both services in place, the pipeline is complete: POST an event, it persists to PostgreSQL, publishes to Kafka, and the consumer picks it up. Time to see it work.

  • If your app is still running from the previous step, stop it with Ctrl+C in the terminal.
  • Make sure your Docker containers are still running. Confirm by running:
docker compose ps
  • You should see threat-postgres and threat-kafka both running and healthy.
  • Start the application by running:
mvn spring-boot:run
  • Wait for the startup logs to finish. You'll see a line like Started ThreatIngestApplication when the app is ready.

Before you send a test event, predict what you expect to see in the terminal logs. Will you see one log line or two? Why?

  • Open a second terminal and POST a test event by running:
curl -X POST http://localhost:8080/api/events \
  -H "Content-Type: application/json" \
  -d '{"sourceIp": "10.0.0.5", "eventType": "PORT_SCAN", "targetSystem": "web-server-01", "rawPayload": "{\"ports\": [22, 80, 443]}"}'
  • Switch back to the terminal running your app and check the logs.

You should see two log lines confirming the full loop:

  • Published event to topic: threat-events with id: <uuid> from the producer.
  • Consumed event from Kafka: <uuid> from the consumer.

Both UUIDs should match. That confirms the event traveled from your REST controller, through PostgreSQL persistence, into Kafka, and back out to the consumer.

Not seeing the consumed log line?

  • Check that threat-kafka is healthy by running docker compose ps. If it shows as unhealthy, restart it with docker compose restart kafka and wait 30 seconds for it to become healthy.
  • Verify that spring.json.trusted.packages in application.yml is set to com.threatpipeline.threatingest.model. Without this, the deserializer rejects the incoming JSON.
  • If you see a serialization error in the logs, confirm the @KafkaListener method parameter type is ThreatEvent (not String).

Why isn't my Kafka consumer receiving events?

✔️ Awesome, I've got everything!

Great. Double check you've saved both files.

ⓧ I'd like to double check the full code

Here is KafkaProducerService.java:

package com.threatpipeline.threatingest.service;

import com.threatpipeline.threatingest.model.ThreatEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaProducerService {

    private static final Logger log = LoggerFactory.getLogger(KafkaProducerService.class);
    private static final String TOPIC = "threat-events";

    private final KafkaTemplate<String, ThreatEvent> kafkaTemplate;

    public KafkaProducerService(KafkaTemplate<String, ThreatEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void publishEvent(ThreatEvent event) {
        kafkaTemplate.send(TOPIC, event.getId().toString(), event);
        log.info("Published event to topic: {} with id: {}", TOPIC, event.getId());
    }
}

Here is KafkaConsumerService.java:

package com.threatpipeline.threatingest.service;

import com.threatpipeline.threatingest.model.ThreatEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

@Service
public class KafkaConsumerService {

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumerService.class);

    @KafkaListener(topics = "threat-events", groupId = "threat-analyzer-group")
    public void consumeEvent(ThreatEvent event) {
        log.info("Consumed event from Kafka: {}", event.getId());
        // Placeholder: analysis logic will be added in the next step
    }
}

Your events now stream through Kafka in real time. Next up, you'll build the threat analysis engine that turns those raw events into actionable alerts with severity classifications.

Implement Threat Analysis and Alert Generation

Your pipeline can ingest events and stream them through Kafka. But raw events are just noise without analysis.

Security teams need actionable alerts with severity levels, not a firehose of undifferentiated logs. In this step, you will build a threat analysis engine that classifies each event and generates enriched alerts.

In this step, get ready to:
  • Create the Severity enum, ThreatAlert entity, and ThreatAlertRepository for persisting enriched alerts.
  • Build ThreatAnalysisService with severity classification rules and a repeat-offender escalation mechanism.
  • Create AlertController to query alerts and verify the full analysis pipeline end-to-end.
Create the alert data model

Before you can classify threats, you need a data model to represent severity levels and the alerts themselves. Your KafkaConsumerService already references ThreatAnalysisService, so the app cannot start until you create the supporting types.

  • Create a new file Severity.java in src/main/java/com/threatpipeline/threatingest/model/ with the severity enum:
package com.threatpipeline.threatingest.model;

public enum Severity {
    LOW,
    MEDIUM,
    HIGH,
    CRITICAL
}

What does this enum represent?

These four severity levels mirror real SIEM alert classifications. LOW captures routine events like single failed logins. CRITICAL flags active attacks that need immediate human response.

  • Create a new file ThreatAlert.java in the same model/ package. Start with the entity declaration and fields:
package com.threatpipeline.threatingest.model;

import jakarta.persistence.*;
import java.time.Instant;
import java.util.UUID;

@Entity
@Table(name = "threat_alerts")
public class ThreatAlert {

    @Id
    @GeneratedValue(strategy = GenerationType.UUID)
    private UUID id;

    @Column(nullable = false)
    private UUID linkedEventId;

    @Enumerated(EnumType.STRING)
    @Column(nullable = false)
    private Severity severity;

    @Column(nullable = false)
    private String description;

    @Column(nullable = false)
    private Instant detectedAt;

    @Column(nullable = false)
    private String ruleTriggered;

What do these fields track?

  • linkedEventId ties each alert back to the raw event that triggered it.
  • severity stores the classification from your rules engine (LOW through CRITICAL).
  • ruleTriggered records which detection rule fired, making alerts auditable.
  • Below the ruleTriggered field, add the final field, constructor, and getter/setter methods to complete the class:
    @Column(nullable = false)
    private String sourceIp;

    public ThreatAlert() {}

    public UUID getId() { return id; }
    public void setId(UUID id) { this.id = id; }
    public UUID getLinkedEventId() { return linkedEventId; }
    public void setLinkedEventId(UUID linkedEventId) { this.linkedEventId = linkedEventId; }
    public Severity getSeverity() { return severity; }
    public void setSeverity(Severity severity) { this.severity = severity; }
    public String getDescription() { return description; }
    public void setDescription(String description) { this.description = description; }
    public Instant getDetectedAt() { return detectedAt; }
    public void setDetectedAt(Instant detectedAt) { this.detectedAt = detectedAt; }
    public String getRuleTriggered() { return ruleTriggered; }
    public void setRuleTriggered(String ruleTriggered) { this.ruleTriggered = ruleTriggered; }
    public String getSourceIp() { return sourceIp; }
    public void setSourceIp(String sourceIp) { this.sourceIp = sourceIp; }
}

Why store sourceIp directly on the alert?

The repeat-offender rule needs to count past alerts by IP address. Storing sourceIp directly on the alert avoids an expensive join back to the events table every time you classify a new event.

  • Create a new file ThreatAlertRepository.java in src/main/java/com/threatpipeline/threatingest/repository/ with these query methods:
package com.threatpipeline.threatingest.repository;

import com.threatpipeline.threatingest.model.Severity;
import com.threatpipeline.threatingest.model.ThreatAlert;
import org.springframework.data.jpa.repository.JpaRepository;
import java.util.List;
import java.util.UUID;

public interface ThreatAlertRepository extends JpaRepository<ThreatAlert, UUID> {
    List<ThreatAlert> findBySeverity(Severity severity);
    long countBySourceIp(String sourceIp);
}

What does countBySourceIp do?

Spring Data JPA auto-generates a SQL COUNT query from the method name. This counts how many prior alerts exist from a given IP address. The repeat-offender rule uses this count to decide whether to escalate severity.

  • Save Severity.java, ThreatAlert.java, and ThreatAlertRepository.java.
  • Confirm your new model files exist by running:
ls src/main/java/com/threatpipeline/threatingest/model/

You should see EventType.java, Severity.java, ThreatAlert.java, and ThreatEvent.java listed.

Missing files in the listing?

Make sure you created the files in the correct directory path. The package com.threatpipeline.threatingest.model maps to src/main/java/com/threatpipeline/threatingest/model/. If you placed them elsewhere, move them now.

Help me find where my Java model files should go.

Build ThreatAnalysisService with severity rules

Your KafkaConsumerService already calls analysisService.analyze(event) on every consumed event. The app cannot start until you provide the actual ThreatAnalysisService bean.

  • Create a new file ThreatAnalysisService.java in src/main/java/com/threatpipeline/threatingest/service/. Start with the class structure and constructor:
package com.threatpipeline.threatingest.service;

import com.threatpipeline.threatingest.model.*;
import com.threatpipeline.threatingest.repository.ThreatAlertRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.time.Instant;
import java.util.List;

@Service
public class ThreatAnalysisService {

    private static final Logger log = LoggerFactory.getLogger(ThreatAnalysisService.class);
    // Escalate to CRITICAL after this many alerts from the same IP
    private static final long REPEAT_OFFENDER_THRESHOLD = 3;

    private final ThreatAlertRepository alertRepository;

    public ThreatAnalysisService(ThreatAlertRepository alertRepository) {
        this.alertRepository = alertRepository;
    }

What is REPEAT_OFFENDER_THRESHOLD?

This constant defines how many prior alerts from the same IP address trigger an escalation to CRITICAL. Once an attacker's IP has generated 3 or more alerts, any new brute-force attempt from that IP is automatically classified as the highest severity.

  • Below the constructor, add the analyze() method that orchestrates classification and alert creation:
    // Orchestrates severity classification and alert persistence
    public ThreatAlert analyze(ThreatEvent event) {
        Severity severity = classifySeverity(event);
        String rule = determineRule(event, severity);

        ThreatAlert alert = new ThreatAlert();
        alert.setLinkedEventId(event.getId());
        alert.setSeverity(severity);
        alert.setDescription(buildDescription(event, severity));
        alert.setDetectedAt(Instant.now());
        alert.setRuleTriggered(rule);
        alert.setSourceIp(event.getSourceIp());

        ThreatAlert saved = alertRepository.save(alert);
        log.info("Generated alert: severity={}, rule={}, sourceIp={}", severity, rule, event.getSourceIp());
        return saved;
    }

What does analyze() do?

  • classifySeverity() applies the rules engine to determine threat level.
  • determineRule() records which detection rule fired for audit purposes.
  • The alert is persisted to PostgreSQL so it can be queried through the REST API.
  • Below analyze(), add the severity classification methods:
    // Maps each event type to a base severity level
    private Severity classifySeverity(ThreatEvent event) {
        return switch (event.getEventType()) {
            case MALWARE_DETECTED -> Severity.CRITICAL;
            case BRUTE_FORCE -> evaluateBruteForce(event.getSourceIp());
            case PORT_SCAN -> Severity.MEDIUM;
            case FAILED_LOGIN -> Severity.LOW;
        };
    }

    // Escalates brute force to CRITICAL if IP is a repeat offender
    private Severity evaluateBruteForce(String sourceIp) {
        long priorAlerts = alertRepository.countBySourceIp(sourceIp);
        if (priorAlerts >= REPEAT_OFFENDER_THRESHOLD) {
            return Severity.CRITICAL;
        }
        return Severity.HIGH;
    }

How do the severity rules work?

  • MALWARE_DETECTED is always CRITICAL. Malware on a system demands immediate response.
  • BRUTE_FORCE starts at HIGH. If the same IP already has 3+ prior alerts, it escalates to CRITICAL.
  • PORT_SCAN is MEDIUM. Port scans are reconnaissance, not active exploitation.
  • FAILED_LOGIN is LOW. A single failed login is usually a typo, not an attack.
  • Below evaluateBruteForce(), add the helper methods and public query method to close the class:
    // Determines which named rule fired for auditing
    private String determineRule(ThreatEvent event, Severity severity) {
        if (event.getEventType() == EventType.BRUTE_FORCE && severity == Severity.CRITICAL) {
            return "REPEAT_OFFENDER_ESCALATION";
        }
        return event.getEventType().name() + "_DETECTION";
    }

    private String buildDescription(ThreatEvent event, Severity severity) {
        return String.format("%s detected from %s targeting %s",
                event.getEventType(), event.getSourceIp(), event.getTargetSystem());
    }

    // Public query method used by AlertController
    public List<ThreatAlert> getAlerts(Severity severity) {
        if (severity != null) {
            return alertRepository.findBySeverity(severity);
        }
        return alertRepository.findAll();
    }
}

What does determineRule() record?

When a brute-force event triggers the repeat-offender escalation, the rule is recorded as REPEAT_OFFENDER_ESCALATION. All other event types record their type name plus _DETECTION (e.g. PORT_SCAN_DETECTION). This makes each alert auditable.

  • Save ThreatAnalysisService.java.

✔️ Awesome, I've got everything!

Great. Confirm the file is saved before moving on.

ⓧ I'd like to double check the full code

Your ThreatAnalysisService.java should contain these four chunks in order:

  • Chunk 1: Package, imports, @Service annotation, logger, REPEAT_OFFENDER_THRESHOLD constant, alertRepository field, and constructor.
  • Chunk 2: The analyze() method that classifies severity, builds the alert, persists it, and logs it.
  • Chunk 3: classifySeverity() with the switch expression and evaluateBruteForce() with the threshold check.
  • Chunk 4: determineRule(), buildDescription(), and getAlerts() closing the class with }.

The analysis engine is wired. Your app should now compile and start successfully.

  • Stop the running app if it is still running (press Ctrl+C in the terminal).
  • Restart the application by running:
mvn spring-boot:run

Before you test, what do you think will happen when you POST a BRUTE_FORCE event now? Will the consumer just log it like before, or will you see something new in the output?

  • Once the app starts successfully, POST a test event by running this in a new terminal:
curl -X POST http://localhost:8080/api/events \
  -H "Content-Type: application/json" \
  -d '{"sourceIp": "192.168.1.100", "eventType": "BRUTE_FORCE", "targetSystem": "auth-server", "rawPayload": "{\"attempts\": 25}"}'

Check your application logs. You should now see both Consumed event from Kafka AND a new line: Generated alert: severity=HIGH, rule=BRUTE_FORCE_DETECTION, sourceIp=192.168.1.100.

App fails to start?

If you see a compilation error about missing symbols, double-check that all new files are in the correct packages and that imports match. The most common issue is a missing import for the Severity enum in ThreatAnalysisService.

If the app starts but you see no "Generated alert" log line, make sure Docker Compose is still running. Kafka must be healthy for the consumer to receive events.

Help me debug why my ThreatAnalysisService isn't generating alerts.

Create AlertController and test the full pipeline

Alerts are being generated and persisted. Now you need a REST endpoint so security teams can query them by severity level.

  • Create a new file AlertController.java in src/main/java/com/threatpipeline/threatingest/controller/:
package com.threatpipeline.threatingest.controller;

import com.threatpipeline.threatingest.model.Severity;
import com.threatpipeline.threatingest.model.ThreatAlert;
import com.threatpipeline.threatingest.service.ThreatAnalysisService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.web.bind.annotation.*;
import java.util.List;

@RestController
@RequestMapping("/api/alerts")
@Tag(name = "Threat Alerts", description = "Query enriched threat alerts with severity classifications")
public class AlertController {

    private final ThreatAnalysisService analysisService;

    public AlertController(ThreatAnalysisService analysisService) {
        this.analysisService = analysisService;
    }

    @GetMapping
    @Operation(summary = "Query threat alerts", description = "Returns all alerts, optionally filtered by severity")
    public List<ThreatAlert> getAlerts(@RequestParam(required = false) Severity severity) {
        return analysisService.getAlerts(severity);
    }
}

What does this controller expose?

GET /api/alerts returns all generated alerts. Adding ?severity=CRITICAL filters to only critical-level threats. The @Tag annotation groups this endpoint under "Threat Alerts" in Swagger UI.

  • Save AlertController.java.
  • Stop the running app (Ctrl+C) and restart it:
mvn spring-boot:run

Now for the full pipeline test. You will POST multiple BRUTE_FORCE events from the same IP and watch severity escalate as the repeat-offender rule kicks in.

Think about this: if the threshold is 3 prior alerts and you POST 4 events from the same IP, which event will be the first to receive CRITICAL severity?

  • POST the first BRUTE_FORCE event by running:
curl -X POST http://localhost:8080/api/events \
  -H "Content-Type: application/json" \
  -d '{"sourceIp": "192.168.1.100", "eventType": "BRUTE_FORCE", "targetSystem": "auth-server", "rawPayload": "{\"attempts\": 50}"}'
  • Run the same curl command three more times (4 total POSTs from the same IP).
  • Query the alerts endpoint to see the severity escalation:
curl http://localhost:8080/api/alerts | python3 -m json.tool

You should see 4 alerts. The first 3 have severity: HIGH with rule BRUTE_FORCE_DETECTION. The 4th has severity: CRITICAL with rule REPEAT_OFFENDER_ESCALATION.

  • Verify you can also filter by severity:
curl http://localhost:8080/api/alerts?severity=CRITICAL | python3 -m json.tool

This should return only the single CRITICAL alert, proving the query filter works.

Seeing an empty array from /api/alerts?

If you restarted the app between posting events and querying alerts, the database was reset. The ddl-auto: create-drop setting drops all tables on startup. POST fresh events after each restart.

If you see events but no alerts, check the app logs for errors in the Kafka consumer. The consumer must successfully deserialize the event and pass it to the analysis service.

Help me debug why my alerts endpoint returns empty.

Your threat analysis engine is live. Raw events flow in, get classified by severity rules, and surface as queryable alerts with full audit trails. Next up, you will add OpenAPI documentation, health checks, structured logging, and tests to make this project production-ready.

Add OpenAPI Docs, Health Checks, Structured Logging, and Tests

Your pipeline works end-to-end, but a project without documentation, observability, and tests signals incomplete engineering to anyone reviewing your code.

In this step, you will add interactive API documentation via Swagger UI, verify health-check endpoints, configure structured JSON logging, and write automated tests that prove the full pipeline works.

In this step, get ready to:
  • Add a custom OpenAPI configuration bean and verify Swagger UI renders your API docs.
  • Verify Spring Actuator health checks and add structured JSON logging with Logback.
  • Write unit tests with Mockito and an integration test with Testcontainers.
Create the OpenAPI configuration bean and verify Swagger UI

Your controllers already have @Tag, @Operation, and @ApiResponse annotations from the previous steps.

To give the generated docs a proper title and description, you need a configuration bean that customises the OpenAPI metadata.

  • Create a new package folder and file at src/main/java/com/threatpipeline/threatingest/config/OpenApiConfig.java and paste the following code:
package com.threatpipeline.threatingest.config;

import io.swagger.v3.oas.models.OpenAPI;
import io.swagger.v3.oas.models.info.Info;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class OpenApiConfig {

    @Bean
    public OpenAPI customOpenAPI() {
        return new OpenAPI()
                .info(new Info()
                        .title("Threat Detection Pipeline API")
                        .version("1.0.0")
                        .description("REST API for ingesting security events, streaming through Kafka, and querying enriched threat alerts"));
    }
}

What does this code do?

The @Configuration class registers a custom OpenAPI bean that Spring picks up at startup.

The Info object sets the title, version, and description that appear at the top of the Swagger UI page. Without this bean, springdoc would generate docs with a generic title.

  • Save the file.
  • Make sure your Docker Compose services are still running. If they stopped, restart them by running:
docker compose up -d
  • Restart the Spring Boot application by running:
mvn spring-boot:run
  • Open your browser and navigate to http://localhost:8080/swagger-ui.html.

You should see an interactive API documentation page titled "Threat Detection Pipeline API" with two tag groups: "Threat Events" and "Threat Alerts".

Swagger UI shows a blank page or 404?

  • Confirm that springdoc-openapi-starter-webmvc-ui version 2.8.17 is in your pom.xml dependencies.
  • Check that your application.yml has the springdoc.swagger-ui.path set to /swagger-ui.html.
  • Make sure the app started without errors in the terminal. Look for "Started ThreatIngestApplication" in the logs.

Help me debug why Swagger UI is not loading at /swagger-ui.html in my Spring Boot 3.5 app with springdoc-openapi.

Verify health checks and add structured JSON logging

Your application.yml already exposes Spring Actuator endpoints with detailed health information.

  • With the app still running, open a new terminal and check the health endpoint by running:
curl http://localhost:8080/actuator/health | python3 -m json.tool

You should see a JSON response with "status": "UP" and nested details for both db (PostgreSQL) and kafka showing that both dependencies are connected and healthy.

Why expose health details?

The management.endpoint.health.show-details=always setting tells Actuator to include connection status for each dependency. In production, you would restrict this to authenticated users, but for development it gives you instant visibility into whether PostgreSQL or Kafka are unreachable.

Now you will add structured JSON logging so that log output can be parsed by monitoring tools.

  • Create the file src/main/resources/logback-spring.xml and paste the following configuration:
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <include resource="org/springframework/boot/logging/logback/defaults.xml"/>

    <springProfile name="!json">
        <include resource="org/springframework/boot/logging/logback/console-appender.xml"/>
        <root level="INFO">
            <appender-ref ref="CONSOLE"/>
        </root>
    </springProfile>

    <springProfile name="json">
        <appender name="JSON_CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
            <encoder class="net.logstash.logback.encoder.LogstashEncoder"/>
        </appender>
        <root level="INFO">
            <appender-ref ref="JSON_CONSOLE"/>
        </root>
    </springProfile>
</configuration>

What does this configuration do?

  • The !json profile (default) uses Spring Boot's standard human-readable console output.
  • The json profile switches to LogstashEncoder, which outputs every log line as a structured JSON object with timestamp, level, message, and context fields.
  • You activate the JSON profile by adding --spring.profiles.active=json to your startup command when deploying to production.
  • Save logback-spring.xml.
  • Stop the running app with Ctrl+C, then restart it to pick up the new logging config by running:
mvn spring-boot:run

You should see the same human-readable log output as before. The JSON format only activates when you explicitly enable the json profile.

  • Stop the app with Ctrl+C. You will not need it running for the next substep.

Seeing a ClassNotFoundException for LogstashEncoder?

  • Confirm that logstash-logback-encoder version 8.0 is in your pom.xml dependencies section.
  • Run mvn clean compile to force Maven to resolve the dependency.

Help me resolve a ClassNotFoundException for LogstashEncoder in my Spring Boot project.

Write unit tests for the threat analysis service

Unit tests verify your severity classification logic in isolation, without needing a database or Kafka broker running.

You will use Mockito to replace the real repository with a mock, so each test focuses purely on the classification rules inside ThreatAnalysisService.

  • Create the test file at src/test/java/com/threatpipeline/threatingest/service/ThreatAnalysisServiceTest.java and paste the imports and class declaration:
package com.threatpipeline.threatingest.service;

import com.threatpipeline.threatingest.model.*;
import com.threatpipeline.threatingest.repository.ThreatAlertRepository;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;

import java.time.Instant;
import java.util.UUID;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.when;

@ExtendWith(MockitoExtension.class)
class ThreatAnalysisServiceTest {

    @Mock
    private ThreatAlertRepository alertRepository;

    private ThreatAnalysisService analysisService;

What does this setup do?

  • @ExtendWith(MockitoExtension.class) enables Mockito annotations without a full Spring context.
  • @Mock creates a fake ThreatAlertRepository that you can program to return specific values.
  • Add the setUp method inside the class, below the field declarations:
    @BeforeEach
    void setUp() {
        analysisService = new ThreatAnalysisService(alertRepository);
        when(alertRepository.save(any(ThreatAlert.class))).thenAnswer(invocation -> {
            ThreatAlert alert = invocation.getArgument(0);
            alert.setId(UUID.randomUUID());
            return alert;
        });
    }

What does setUp do?

The setUp method creates a real ThreatAnalysisService with the mock injected, and stubs the save() method to simulate database persistence by returning the alert with a generated ID.

  • Add the first three test methods inside the class, below the setUp method:
    @Test
    void analyzeMalwareDetected_returnsCritical() {
        ThreatEvent event = createEvent(EventType.MALWARE_DETECTED);

        ThreatAlert alert = analysisService.analyze(event);

        assertThat(alert.getSeverity()).isEqualTo(Severity.CRITICAL);
        assertThat(alert.getRuleTriggered()).isEqualTo("MALWARE_DETECTED_DETECTION");
    }

    @Test
    void analyzeBruteForce_returnsHigh() {
        when(alertRepository.countBySourceIp("10.0.0.1")).thenReturn(0L);
        ThreatEvent event = createEvent(EventType.BRUTE_FORCE);

        ThreatAlert alert = analysisService.analyze(event);

        assertThat(alert.getSeverity()).isEqualTo(Severity.HIGH);
    }

    @Test
    void analyzeBruteForceRepeatOffender_returnsCritical() {
        when(alertRepository.countBySourceIp("10.0.0.1")).thenReturn(3L);
        ThreatEvent event = createEvent(EventType.BRUTE_FORCE);

        ThreatAlert alert = analysisService.analyze(event);

        assertThat(alert.getSeverity()).isEqualTo(Severity.CRITICAL);
        assertThat(alert.getRuleTriggered()).isEqualTo("REPEAT_OFFENDER_ESCALATION");
    }

What do these tests verify?

  • MALWARE_DETECTED always classifies as CRITICAL regardless of history.
  • BRUTE_FORCE with no prior alerts from that IP classifies as HIGH.
  • BRUTE_FORCE from a repeat offender (3+ prior alerts) escalates to CRITICAL with the REPEAT_OFFENDER_ESCALATION rule.
  • Add the remaining two test methods and the helper method, still inside the class:
    @Test
    void analyzePortScan_returnsMedium() {
        ThreatEvent event = createEvent(EventType.PORT_SCAN);

        ThreatAlert alert = analysisService.analyze(event);

        assertThat(alert.getSeverity()).isEqualTo(Severity.MEDIUM);
    }

    @Test
    void analyzeFailedLogin_returnsLow() {
        ThreatEvent event = createEvent(EventType.FAILED_LOGIN);

        ThreatAlert alert = analysisService.analyze(event);

        assertThat(alert.getSeverity()).isEqualTo(Severity.LOW);
    }

    // Helper to create a test event with common defaults
    private ThreatEvent createEvent(EventType type) {
        ThreatEvent event = new ThreatEvent();
        event.setId(UUID.randomUUID());
        event.setSourceIp("10.0.0.1");
        event.setEventType(type);
        event.setTargetSystem("prod-server-01");
        event.setTimestamp(Instant.now());
        return event;
    }

What do these tests verify?

  • PORT_SCAN classifies as MEDIUM severity.
  • FAILED_LOGIN classifies as LOW severity.
  • The createEvent helper standardises the test setup so each test only varies the event type.
  • Save ThreatAnalysisServiceTest.java.
  • Run just the unit tests to confirm they pass by running:
mvn test -Dtest=ThreatAnalysisServiceTest

You should see all 5 tests pass with a "BUILD SUCCESS" message.

Tests failing with compilation errors?

  • Make sure the closing brace for the class is present at the end of the file.
  • Check that all test methods and the helper method are inside the class body, not outside it.
  • Verify the package declaration matches the directory path exactly.

Help me fix compilation errors in my Mockito unit test for ThreatAnalysisService.

Now you will write an integration test that boots the full application with real PostgreSQL and Kafka containers to verify the complete pipeline end-to-end.

  • Create the integration test file at src/test/java/com/threatpipeline/threatingest/ThreatPipelineIntegrationTest.java and paste the imports:
package com.threatpipeline.threatingest;

import com.threatpipeline.threatingest.dto.CreateEventRequest;
import com.threatpipeline.threatingest.model.EventType;
import com.threatpipeline.threatingest.model.ThreatAlert;
import com.threatpipeline.threatingest.repository.ThreatAlertRepository;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.client.TestRestTemplate;
import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

import java.util.List;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

What are Testcontainers?

Testcontainers spins up real Docker containers for PostgreSQL and Kafka during your test run. The @ServiceConnection annotation tells Spring Boot to auto-configure the datasource and Kafka connection to point at these ephemeral containers instead of your Docker Compose services.

This means your tests are fully isolated and repeatable on any machine with Docker installed.

  • Add the class body with the container definitions and autowired fields below the imports:
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@Testcontainers
class ThreatPipelineIntegrationTest {

    @Container
    @ServiceConnection
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>(
            DockerImageName.parse("postgres:16"));

    @Container
    @ServiceConnection
    static KafkaContainer kafka = new KafkaContainer(
            DockerImageName.parse("confluentinc/cp-kafka:7.7.1"));

    @Autowired
    private TestRestTemplate restTemplate;

    @Autowired
    private ThreatAlertRepository alertRepository;
  • Add the test method that exercises the full pipeline, then close the class:
    @Test
    void fullPipeline_eventIngestionGeneratesAlert() {
        // Build a BRUTE_FORCE event request
        CreateEventRequest request = new CreateEventRequest();
        request.setSourceIp("192.168.1.50");
        request.setEventType(EventType.BRUTE_FORCE);
        request.setTargetSystem("auth-server");
        request.setRawPayload("{\"attempts\": 50}");

        // POST the event to the REST API
        ResponseEntity<String> response = restTemplate.postForEntity(
                "/api/events", request, String.class);

        assertThat(response.getStatusCode()).isEqualTo(HttpStatus.CREATED);

        // Wait for Kafka consumer to process and generate an alert
        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
            List<ThreatAlert> alerts = alertRepository.findAll();
            assertThat(alerts).isNotEmpty();
            assertThat(alerts.get(0).getSourceIp()).isEqualTo("192.168.1.50");
        });
    }
}

What does this test prove?

  • The test POSTs a BRUTE_FORCE event to /api/events and asserts a 201 response.
  • It then waits up to 10 seconds for the Kafka consumer to pick up the event and generate an alert in the database.
  • If the alert appears with the correct sourceIp, the entire pipeline (REST to persist to Kafka to consume to analyze to alert) is working.
  • Save ThreatPipelineIntegrationTest.java.

Before you run the full test suite, what do you think will happen? The integration test needs to pull Docker images and boot containers, so predict whether it will be fast or slow.

  • Run the full test suite by running:
mvn test

The first run may take a minute as Testcontainers pulls the postgres:16 and confluentinc/cp-kafka:7.7.1 images. You should see all tests pass and a "BUILD SUCCESS" message at the end.

Integration test timing out or failing to start containers?

  • Make sure Docker Desktop is running. Testcontainers needs the Docker daemon to create containers.
  • If you see connection refused errors, check that no other process is blocking Docker's socket.
  • On the first run, image pulls can take several minutes on slow connections. Wait for the full download before assuming failure.

Help me debug why my Testcontainers integration test is failing to start PostgreSQL or Kafka containers.

✔️ Awesome, I've got everything!

Great. Double check you have saved both test files and that mvn test shows BUILD SUCCESS.

ⓧ I'd like to double check the full code

Since the test files exceed 30 lines each, confirm your files contain these sections in order:

ThreatAnalysisServiceTest.java:

  • Package declaration and imports (assertThat, Mockito, model classes).
  • @ExtendWith(MockitoExtension.class) class with @Mock alertRepository and @BeforeEach setUp.
  • Five @Test methods: analyzeMalwareDetected_returnsCritical, analyzeBruteForce_returnsHigh, analyzeBruteForceRepeatOffender_returnsCritical, analyzePortScan_returnsMedium, analyzeFailedLogin_returnsLow.
  • The private createEvent helper method.

ThreatPipelineIntegrationTest.java:

  • Package declaration and imports (Testcontainers, TestRestTemplate, Awaitility).
  • @SpringBootTest with RANDOM_PORT and @Testcontainers annotations on the class.
  • Static @Container @ServiceConnection fields for PostgreSQLContainer (postgres:16) and KafkaContainer (confluentinc/cp-kafka:7.7.1).
  • @Autowired TestRestTemplate and ThreatAlertRepository fields.
  • The fullPipeline_eventIngestionGeneratesAlert test method with POST, assert 201, and await alert.

For OpenApiConfig.java, confirm it matches this:

package com.threatpipeline.threatingest.config;

import io.swagger.v3.oas.models.OpenAPI;
import io.swagger.v3.oas.models.info.Info;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class OpenApiConfig {

    @Bean
    public OpenAPI customOpenAPI() {
        return new OpenAPI()
                .info(new Info()
                        .title("Threat Detection Pipeline API")
                        .version("1.0.0")
                        .description("REST API for ingesting security events, streaming through Kafka, and querying enriched threat alerts"));
    }
}

Your threat detection pipeline is now fully documented, observable, and tested. You have interactive API docs, health monitoring, structured logs ready for production, and automated tests that prove the pipeline works from end to end. Next up, you will tackle the Secret Mission and decompose this application into independently deployable microservices.

Secret mission

Extract the Analyzer into a Second Microservice

In production SIEM platforms, event ingestion and threat analysis are separate services. They scale independently and deploy on their own schedules. In this mission, you will decompose your monolith into two microservices communicating through Kafka, proving you can architect a real distributed pipeline.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally via Docker, so there are no ongoing costs.

Resources you used:

  • Docker containers: threat-postgres, alerts-postgres, threat-kafka.
  • Docker volumes: pgdata, alertsdata, kafkadata.
  • Local project files: the threat-ingest folder on your Desktop (includes the threat-analyzer subfolder).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the threat pipeline.

  • Your Docker containers continue running in the background and will restart if Docker Desktop restarts.
  • Your project files remain on your Desktop.
  • PostgreSQL and Kafka will consume some system memory while running.

Pause - I'll come back to this later

Shut down running containers to free up memory. Your data volumes and project files stay intact so you can resume later.

  • Stop the Spring Boot applications if they are still running by pressing Ctrl+C in each terminal.
  • Stop and remove the Docker containers by running this command from your project folder:
docker compose down
  • Your data volumes (pgdata, alertsdata, kafkadata) are preserved. When you return, run docker compose up -d to bring everything back.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes all containers, volumes, and project files permanently.

  • Stop the Spring Boot applications if they are still running by pressing Ctrl+C in each terminal.
  • Stop and remove all Docker containers and volumes by running this command from your project folder:
docker compose down -v
  • Delete the project folder from your Desktop by running:
rm -rf ~/Desktop/threat-ingest
  • Confirm the folder is gone by running:
ls ~/Desktop | grep threat
  • You should see no output, confirming the project files are deleted.

Nice Work!

Nice Work!

You just built a production-grade SIEM threat detection pipeline from scratch. Security events flow in through your REST API, stream through Kafka, get analyzed for severity, and surface as queryable alerts.

You've learned how to:

  • Build a threat event ingestion REST API with Jakarta Validation and Spring Data JPA persistence, then stream events through Apache Kafka as a real-time event backbone.
  • Implement a threat analysis engine with severity classification rules and repeat-offender escalation, containerized alongside PostgreSQL and Kafka using Docker Compose.
  • Add production-readiness features including OpenAPI documentation with Swagger UI, Spring Actuator health checks, structured JSON logging, and automated tests with Mockito and Testcontainers.
  • Secret Mission: Decomposed the monolith into two independent microservices communicating via Kafka, proving you can break apart a service for independent scaling and deployment.

Ready to quiz yourself?