Real-Time Security Audit Pipeline

Build a Kafka Streams pipeline that detects credential-stuffing attacks in real time.

Introduction

30 Second Summary

Every time you log into a website, that action generates a security event. When attackers try thousands of passwords across hundreds of accounts in seconds, organizations need to detect that pattern instantly. Not minutes later. Not after a report runs overnight.

In this project, you will build a real-time security audit pipeline that streams login events through Apache Kafka, detects credential-stuffing attacks using Kafka Streams, and visualizes threats in Kibana.

What You'll Build

A security event pipeline running locally in Docker where a Spring Boot application produces audit events to Kafka, a Kafka Streams topology detects brute-force attacks in real time, and the ELK stack (Elasticsearch, Logstash, Kibana) indexes and visualizes everything.

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

  • A running Kafka broker in KRaft mode with an understanding of how it replaces ZooKeeper's coordination role.
  • A Kafka Streams application that detects credential-stuffing attacks (20+ failed logins across 10+ accounts from one IP within 60 seconds) and produces alert events.
  • A full ELK pipeline consuming from Kafka that lets you search audit logs and visualize attack patterns in Kibana dashboards.
  • Secret Mission: Add geo-enrichment to map attacker IPs to countries and build a Kibana map visualization showing global attack origins.

Are there any prerequisites?

This project assumes you have Java and Spring Boot fundamentals (creating classes, annotations, dependency injection). You should also have basic Docker knowledge (running containers, using docker compose up and docker compose down).

You do not need prior experience with Kafka, Elasticsearch, or stream processing. This project teaches those from scratch.

Set Up Docker Compose and Project Scaffold

Before you can detect security threats in real time, you need infrastructure to carry the event data. This project uses four services working together: a Kafka broker to stream events, Elasticsearch to store them, Logstash to bridge the two, and Kibana to visualize them.

In this step, you will define all four services in a single Docker Compose file and scaffold the Spring Boot application that produces and processes security events.

In this step, get ready to:
  • Create a Docker Compose file running Kafka in KRaft mode, Elasticsearch, Logstash, and Kibana.
  • Scaffold a Spring Boot Maven project with Kafka and web dependencies.
  • Verify infrastructure health and create the Kafka topics your pipeline needs.
Create the Docker Compose file with Kafka in KRaft mode

Kafka 4.0 and above removed ZooKeeper entirely. Kafka now manages its own cluster metadata using a built-in consensus protocol called KRaft. Your Docker Compose file configures a single node that acts as both broker and controller.

  • Move to your Desktop by running this command:
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir security-audit-pipeline
cd security-audit-pipeline
  • Create the Logstash pipeline directory (Logstash needs this mount point to start):
mkdir -p logstash/pipeline
  • Confirm the directories exist by running:
ls

You should see logstash in the output.

  • Create the docker-compose.yml file by copying the following code into a new file called docker-compose.yml in your security-audit-pipeline folder:
version: '3.8'
services:
  kafka:
    image: apache/kafka:4.3.0
    container_name: kafka
    ports:
      - '9092:9092'
    environment:
      # KAFKA_PROCESS_ROLES replaces the external ZooKeeper ensemble — this node acts as both broker and controller
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      # KAFKA_CONTROLLER_QUORUM_VOTERS replaces zookeeper.connect — lists all controller nodes in the KRaft quorum
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      # KAFKA_LOG_DIRS replaces ZooKeeper-managed broker metadata storage
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    volumes:
      - kafka-data:/var/lib/kafka/data

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:9.4.2
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - ES_JAVA_OPTS=-Xms1g -Xmx1g
    ports:
      - '9200:9200'
    volumes:
      - es-data:/usr/share/elasticsearch/data

  logstash:
    image: docker.elastic.co/logstash/logstash:9.4.2
    container_name: logstash
    ports:
      - '5044:5044'
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline
    depends_on:
      - elasticsearch
      - kafka

  kibana:
    image: docker.elastic.co/kibana/kibana:9.4.2
    container_name: kibana
    ports:
      - '5601:5601'
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    depends_on:
      - elasticsearch

volumes:
  kafka-data:
  es-data:

What does this configuration do?

The Kafka service runs in KRaft combined mode. Three comments in the file highlight what each KRaft setting replaces from the old ZooKeeper architecture:

  • KAFKA_PROCESS_ROLES: broker,controller makes this single node handle both client requests and cluster coordination. Previously, a separate ZooKeeper ensemble handled coordination.
  • KAFKA_CONTROLLER_QUORUM_VOTERS lists the controller nodes that vote on metadata changes. This replaces the old zookeeper.connect property.
  • KAFKA_LOG_DIRS is where Kafka persists its own metadata now, replacing the ZooKeeper znodes that used to store broker state.

Elasticsearch runs in single-node discovery mode with security disabled to avoid certificate setup. Kibana connects to Elasticsearch over the Docker network using the container hostname.

Why KRaft instead of ZooKeeper?

Apache Kafka 4.0 removed ZooKeeper support entirely. KRaft is not optional. It embeds the metadata management directly into the Kafka process, reducing operational complexity from two separate distributed systems to one.

For learners who have seen older Kafka tutorials: there is no zookeeper service in this Compose file. That is the point. Kafka handles everything internally now.

  • Save docker-compose.yml.

Not sure where to create the file?

The file goes in the root of your security-audit-pipeline folder on the Desktop. If you are using VS Code or IntelliJ, open that folder and create the file from the editor sidebar. If you prefer the terminal, use touch docker-compose.yml then paste the content.

Help me create docker-compose.yml in the right location.

Scaffold the Spring Boot Maven project

Your Spring Boot application will produce audit events and run the Kafka Streams detection topology. You need five dependencies: the core starter, web endpoints, Spring Kafka integration, the Kafka Streams library, and Jackson for JSON serialization.

  • Create the Maven project directory structure by running:
mkdir -p src/main/java/com/nextwork/securityaudit
mkdir -p src/main/resources
  • Confirm the structure exists by running:
ls src/main/java/com/nextwork/securityaudit

The command should complete without error. An empty directory listing is expected at this point.

  • Create a pom.xml file in your security-audit-pipeline folder with the following content:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.4.13</version>
  </parent>
  <groupId>com.nextwork</groupId>
  <artifactId>security-audit-pipeline</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <properties>
    <java.version>17</java.version>
  </properties>
  <dependencies>
    <!-- Core Spring Boot auto-configuration -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <!-- Embedded web server for REST endpoints -->
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!-- Spring Kafka integration (KafkaTemplate, listeners) -->
    <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- Kafka Streams DSL for real-time processing -->
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-streams</artifactId>
    </dependency>
    <!-- JSON serialization for audit events -->
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

What do these dependencies do?

  • spring-boot-starter provides core auto-configuration and logging.
  • spring-boot-starter-web adds an embedded Tomcat server so you can expose REST endpoints for submitting audit events.
  • spring-kafka gives you KafkaTemplate for producing messages and listener annotations for consuming them.
  • kafka-streams is the Kafka Streams DSL library for building stateful stream processing topologies (your brute-force detector).
  • jackson-databind handles converting Java objects to JSON and back when sending events through Kafka.
  • Create the application.properties file at src/main/resources/application.properties with the following content:
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.streams.application-id=security-audit-streams
spring.kafka.streams.default-key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.default-value-serde=org.springframework.kafka.support.serializer.JsonSerde
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
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=com.nextwork.*

What do these properties configure?

The bootstrap-servers property tells the Spring Boot app where to find Kafka. Since your app runs on the host machine (not inside Docker), it connects via localhost:9092.

The streams.application-id doubles as the consumer group ID for your Kafka Streams topology. This is what lets Streams and Logstash consume the same topic independently. Each has its own group ID, so Kafka delivers every message to both.

The serializer and deserializer settings ensure all messages are sent and received as JSON, with keys as plain strings. The trusted packages setting allows Jackson to deserialize your project's event classes.

  • Save both pom.xml and application.properties.
  • Verify Maven can resolve the project by running:
mvn validate

You should see BUILD SUCCESS in the output. This confirms Maven recognizes your pom and can reach the parent starter.

Seeing BUILD FAILURE?

  • Make sure pom.xml is in the root of security-audit-pipeline (the same folder where docker-compose.yml lives).
  • Confirm you have internet connectivity. Maven downloads the Spring Boot parent POM on first run.
  • Check that JDK 17 is your active Java version by running java -version.

Help me fix my Maven BUILD FAILURE error.

✔️ Awesome, I've got everything!

Great. Double check you have saved both files before moving on.

ⓧ I'd like to double check the full code

Here is the complete pom.xml:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.4.13</version>
  </parent>
  <groupId>com.nextwork</groupId>
  <artifactId>security-audit-pipeline</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <properties>
    <java.version>17</java.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-streams</artifactId>
    </dependency>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-maven-plugin</artifactId>
      </plugin>
    </plugins>
  </build>
</project>

And the complete application.properties:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.streams.application-id=security-audit-streams
spring.kafka.streams.default-key-serde=org.apache.kafka.common.serialization.Serdes$StringSerde
spring.kafka.streams.default-value-serde=org.springframework.kafka.support.serializer.JsonSerde
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
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=com.nextwork.*
Start the infrastructure and create Kafka topics

With the Docker Compose file in place, you can bring up all four services at once. After they start, you will create the two Kafka topics your pipeline uses: one for raw audit events and one for security alerts.

  • Start all containers in detached mode by running:
docker compose up -d

What does this command do?

The -d flag runs containers in the background so your terminal stays free. Docker Compose reads docker-compose.yml, pulls the images if they are not cached locally, and starts all four services on a shared network.

The first run may take a few minutes while Docker downloads the Kafka, Elasticsearch, Logstash, and Kibana images (several GB combined).

  • Verify all containers are running by executing:
docker compose ps

You should see four containers (kafka, elasticsearch, logstash, kibana) all with a status of Up.

Container exiting or restarting?

  • Run docker compose logs kafka (or replace kafka with the failing service name) to see error output.
  • Elasticsearch requires at least 2 GB of free RAM. If it crashes, close other applications or increase Docker Desktop's memory allocation in Settings > Resources.
  • If Logstash exits because no pipeline config exists yet, that is expected. It will restart once you add the pipeline in a later step.

Help me fix my Docker containers that keep restarting.

  • Create the audit-events topic by running:
docker exec --workdir /opt/kafka/bin/ kafka ./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic audit-events
  • Create the security-alerts topic by running:
docker exec --workdir /opt/kafka/bin/ kafka ./kafka-topics.sh --bootstrap-server localhost:9092 --create --topic security-alerts

You should see Created topic audit-events. and Created topic security-alerts. after each command.

Why create topics manually?

The Docker Compose file sets KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'. In production, auto-creating topics on first message is dangerous because typos in topic names silently create garbage topics. Explicit creation forces you to declare your data contracts up front.

  • Confirm Elasticsearch is healthy by running:
curl localhost:9200

You should see a JSON response containing the Elasticsearch cluster name and version number (9.4.2).

Connection refused on port 9200?

  • Elasticsearch may still be starting. Wait 30 seconds and retry the curl command.
  • Run docker compose logs elasticsearch to check for out-of-memory errors or startup failures.
  • Make sure port 9200 is not already in use by another process on your machine.

Help me troubleshoot Elasticsearch not responding.

✔️ Awesome, I've got everything!

Great. Double check you have saved docker-compose.yml and that all containers are running.

ⓧ I'd like to double check the full code

Here is the complete docker-compose.yml:

version: '3.8'
services:
  kafka:
    image: apache/kafka:4.3.0
    container_name: kafka
    ports:
      - '9092:9092'
    environment:
      # KAFKA_PROCESS_ROLES replaces the external ZooKeeper ensemble — this node acts as both broker and controller
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_NODE_ID: 1
      # KAFKA_CONTROLLER_QUORUM_VOTERS replaces zookeeper.connect — lists all controller nodes in the KRaft quorum
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      # KAFKA_LOG_DIRS replaces ZooKeeper-managed broker metadata storage
      KAFKA_LOG_DIRS: /var/lib/kafka/data
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'
    volumes:
      - kafka-data:/var/lib/kafka/data

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:9.4.2
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - ES_JAVA_OPTS=-Xms1g -Xmx1g
    ports:
      - '9200:9200'
    volumes:
      - es-data:/usr/share/elasticsearch/data

  logstash:
    image: docker.elastic.co/logstash/logstash:9.4.2
    container_name: logstash
    ports:
      - '5044:5044'
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline
    depends_on:
      - elasticsearch
      - kafka

  kibana:
    image: docker.elastic.co/kibana/kibana:9.4.2
    container_name: kibana
    ports:
      - '5601:5601'
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
    depends_on:
      - elasticsearch

volumes:
  kafka-data:
  es-data:

Your infrastructure is live. Next up, you will build the audit event data model and a Kafka producer that streams security events into the audit-events topic.

Build the Audit Event Producer

Your Kafka broker and ELK stack are running in Docker, and both topics are ready to receive messages. Now you need something to actually produce events into the pipeline.

In this step, you will build the data model that represents a security audit event and wire up a Kafka producer that publishes those events to the audit-events topic. You will also expose a REST endpoint so events can be submitted over HTTP and verified end-to-end.

In this step, get ready to:
  • Define the AuditEvent data model and a custom Kafka serializer.
  • Build the AuditEventProducer service that keys events by source IP.
  • Create a REST endpoint and test the full produce-consume cycle.
Define the AuditEvent data model

Every event in the pipeline needs a consistent structure so downstream consumers know exactly what fields to expect. You will model this as a Java record with an enum for the event types.

  • Create the package directory src/main/java/com/nextwork/audit/model inside your project.
  • Create a file called EventType.java in that directory with the following code:
package com.nextwork.audit.model;

public enum EventType {
    LOGIN_SUCCESS,
    LOGIN_FAILED,
    PERMISSION_CHANGE,
    BILLING_UPDATE
}

What does this enum do?

EventType constrains the kinds of security events your pipeline can carry. Using an enum instead of a plain string means the compiler catches typos and the JSON serializer produces predictable values.

  • In the same model package, create a file called AuditEvent.java with the following code:
package com.nextwork.audit.model;

import java.time.Instant;
import java.util.Map;

public record AuditEvent(
    String eventId,
    EventType eventType,
    String userId,
    String sourceIp,
    Instant timestamp,
    Map<String, String> metadata
) {}

What does this record do?

A Java record gives you an immutable data carrier with auto-generated equals(), hashCode(), and toString() methods.

  • eventId uniquely identifies each event for deduplication.
  • sourceIp is the field you will use as the Kafka message key later.
  • timestamp uses Instant for precise, timezone-agnostic event timing.
  • metadata is a flexible map for extra context like user-agent or geo data.
  • Save both files.
Create the custom JSON serializer

Kafka stores messages as byte arrays. You need a serializer that converts your AuditEvent record into JSON bytes before sending it to the broker.

  • Create the package directory src/main/java/com/nextwork/audit/serializer inside your project.
  • Create a file called AuditEventSerializer.java in that directory with the following code:
package com.nextwork.audit.serializer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.nextwork.audit.model.AuditEvent;
import org.apache.kafka.common.serialization.Serializer;

public class AuditEventSerializer implements Serializer<AuditEvent> {
    private final ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());

    @Override
    public byte[] serialize(String topic, AuditEvent data) {
        if (data == null) return null;
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("Error serializing AuditEvent", e);
        }
    }
}

What does this serializer do?

This class implements Kafka's Serializer interface so Kafka knows how to convert an AuditEvent into bytes for the wire.

  • JavaTimeModule teaches Jackson how to serialize Instant fields into ISO-8601 strings. Without it, the timestamp would fail to serialize.
  • The jackson-datatype-jsr310 library is already on your classpath because Spring Boot's web starter includes it transitively.
  • Save AuditEventSerializer.java.
  • Confirm you now have three new files: EventType.java, AuditEvent.java, and AuditEventSerializer.java.

Import errors on JavaTimeModule?

If your IDE shows a red underline on com.fasterxml.jackson.datatype.jsr310.JavaTimeModule, make sure you have run a Maven reload so the transitive dependencies resolve.

In IntelliJ, right-click your pom.xml and select Maven then Reload Project. In VS Code, the Java extension reloads automatically when you save the POM.

help me fix JavaTimeModule import errors in my Spring Boot Kafka project

Build the AuditEventProducer service

With the data model in place, you need a service that actually sends events to Kafka. The key design decision here is what to use as the Kafka message key.

  • Create the package directory src/main/java/com/nextwork/audit/producer inside your project.
  • Create a file called AuditEventProducer.java in that directory with the following code:
package com.nextwork.audit.producer;

import com.nextwork.audit.model.AuditEvent;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class AuditEventProducer {
    private static final String TOPIC = "audit-events";
    private final KafkaTemplate<String, AuditEvent> kafkaTemplate;

    public AuditEventProducer(KafkaTemplate<String, AuditEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // Keying by sourceIp ensures all events from the same IP land on the same partition,
    // enabling the downstream Kafka Streams topology to do per-IP windowed aggregation
    // without a costly repartition step.
    public void send(AuditEvent event) {
        kafkaTemplate.send(TOPIC, event.sourceIp(), event);
    }
}

Why key messages by source IP?

Kafka guarantees ordering within a partition. When you key messages by sourceIp, Kafka hashes the key and routes all events from the same IP to the same partition.

In Step 3, you will build a Kafka Streams topology that aggregates failed logins per IP using a sliding window. Because all events for a given IP are already on one partition, Streams can aggregate them locally without an expensive repartition operation.

  • Save AuditEventProducer.java.

Red underline on KafkaTemplate?

Make sure your pom.xml includes the spring-kafka dependency and that Maven has resolved it. Run a Maven reload if the import does not resolve.

help me resolve KafkaTemplate import issues in my Spring Boot project

Create the REST endpoint and test the pipeline

You now have a producer service, but no way to trigger it from outside the application. A simple REST controller lets you POST audit events over HTTP and verify they land in Kafka.

  • Create the package directory src/main/java/com/nextwork/audit/controller inside your project.
  • Create a file called AuditController.java in that directory with the following code:
package com.nextwork.audit.controller;

import com.nextwork.audit.model.AuditEvent;
import com.nextwork.audit.producer.AuditEventProducer;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/audit")
public class AuditController {
    private final AuditEventProducer producer;

    public AuditController(AuditEventProducer producer) {
        this.producer = producer;
    }

    @PostMapping
    public ResponseEntity<String> publishEvent(@RequestBody AuditEvent event) {
        producer.send(event);
        return ResponseEntity.ok("Event published: " + event.eventId());
    }
}

What does this controller do?

  • @RestController tells Spring this class handles HTTP requests and returns JSON responses.
  • @RequestMapping("/api/audit") maps all endpoints in this class under the /api/audit path.
  • @PostMapping on publishEvent means a POST request to /api/audit triggers the method.
  • @RequestBody deserializes the incoming JSON into an AuditEvent record automatically.
  • Save AuditController.java.
  • Make sure your Docker containers are still running, then start the Spring Boot application by running:
./mvnw spring-boot:run
  • Wait until you see Started in the console output, then open a second terminal window.
  • Send a test audit event to the REST endpoint by running:
curl -X POST http://localhost:8080/api/audit \
  -H "Content-Type: application/json" \
  -d '{
    "eventId": "evt-001",
    "eventType": "LOGIN_FAILED",
    "userId": "user-42",
    "sourceIp": "203.0.113.42",
    "timestamp": "2026-06-23T10:15:00Z",
    "metadata": {"userAgent": "Mozilla/5.0"}
  }'
  • You should see a response: Event published: evt-001.
  • Verify the event landed in Kafka by running the console consumer inside the container:
docker exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic audit-events \
  --from-beginning
  • You should see a JSON object containing eventId, eventType, userId, and the other fields you sent.
  • Press Ctrl+C to stop the consumer.

Not seeing the event in the consumer?

Make sure the Spring Boot application started without errors. Check the terminal running ./mvnw spring-boot:run for stack traces.

Verify the Kafka container is running by executing docker ps and confirming you see the kafka container. If it is not running, restart your infrastructure with docker compose up -d.

help me debug why my audit event is not appearing in the Kafka console consumer

✔️ Awesome, I've got everything!

Great. Double check you have saved all five new files before moving on.

ⓧ I'd like to double check the full code

Here is the complete code for each file created in this step.

package com.nextwork.audit.model;

public enum EventType {
    LOGIN_SUCCESS,
    LOGIN_FAILED,
    PERMISSION_CHANGE,
    BILLING_UPDATE
}
package com.nextwork.audit.model;

import java.time.Instant;
import java.util.Map;

public record AuditEvent(
    String eventId,
    EventType eventType,
    String userId,
    String sourceIp,
    Instant timestamp,
    Map<String, String> metadata
) {}
package com.nextwork.audit.serializer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.nextwork.audit.model.AuditEvent;
import org.apache.kafka.common.serialization.Serializer;

public class AuditEventSerializer implements Serializer<AuditEvent> {
    private final ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());

    @Override
    public byte[] serialize(String topic, AuditEvent data) {
        if (data == null) return null;
        try {
            return objectMapper.writeValueAsBytes(data);
        } catch (Exception e) {
            throw new RuntimeException("Error serializing AuditEvent", e);
        }
    }
}
package com.nextwork.audit.producer;

import com.nextwork.audit.model.AuditEvent;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class AuditEventProducer {
    private static final String TOPIC = "audit-events";
    private final KafkaTemplate<String, AuditEvent> kafkaTemplate;

    public AuditEventProducer(KafkaTemplate<String, AuditEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    // Keying by sourceIp ensures all events from the same IP land on the same partition,
    // enabling the downstream Kafka Streams topology to do per-IP windowed aggregation
    // without a costly repartition step.
    public void send(AuditEvent event) {
        kafkaTemplate.send(TOPIC, event.sourceIp(), event);
    }
}
package com.nextwork.audit.controller;

import com.nextwork.audit.model.AuditEvent;
import com.nextwork.audit.producer.AuditEventProducer;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/api/audit")
public class AuditController {
    private final AuditEventProducer producer;

    public AuditController(AuditEventProducer producer) {
        this.producer = producer;
    }

    @PostMapping
    public ResponseEntity<String> publishEvent(@RequestBody AuditEvent event) {
        producer.send(event);
        return ResponseEntity.ok("Event published: " + event.eventId());
    }
}

Your audit event producer is live and verified. Next up, you will build the Kafka Streams topology that watches the audit-events topic in real time and detects credential-stuffing attacks using sliding windows.

Build the Brute-Force Detection Topology

Your audit event producer is publishing structured events to Kafka, keyed by source IP. Now you need something watching that stream in real time, detecting patterns that humans would miss.

Credential-stuffing attacks follow a distinctive signature: a single IP rapidly fails login attempts across many different accounts. In this step, you will build a Kafka Streams topology that detects this pattern using a 60-second sliding window and compound threshold logic.

In this step, get ready to:
  • Define the aggregation state model and alert record.
  • Build a Kafka Streams topology with windowed aggregation.
  • Add threshold detection that produces alerts to a dedicated topic.
Define the aggregation state and alert models

A simple count of failed logins per IP is not enough to detect credential stuffing. You also need to track how many distinct accounts were targeted. This requires a custom state object that holds both metrics simultaneously.

  • Create a new file called AttackAggregator.java in src/main/java/com/nextwork/audit/model/ by pasting the following code:
package com.nextwork.audit.model;

import java.util.HashSet;
import java.util.Set;

public class AttackAggregator {
    private int failedCount;
    private Set<String> uniqueUserIds;

    public AttackAggregator() {
        this.failedCount = 0;
        this.uniqueUserIds = new HashSet<>();
    }

    // Increments the failure count and tracks the targeted user
    public AttackAggregator increment(String userId) {
        this.failedCount++;
        this.uniqueUserIds.add(userId);
        return this;
    }

    public int getFailedCount() { return failedCount; }
    public Set<String> getUniqueUserIds() { return uniqueUserIds; }
    public void setFailedCount(int failedCount) { this.failedCount = failedCount; }
    public void setUniqueUserIds(Set<String> uniqueUserIds) { this.uniqueUserIds = uniqueUserIds; }
}

What does this code do?

AttackAggregator is a stateful container used inside the Kafka Streams aggregation. Each time a failed login arrives for an IP, the increment() method bumps the failure count and adds the userId to a Set. Because a Set only stores unique values, duplicate userIds are automatically ignored.

The getters and setters are required for JSON serialization. Kafka Streams needs to serialize this object into its internal state store and deserialize it back on every window evaluation.

  • Save AttackAggregator.java.

Next, you need a record type to represent the alert that gets produced when thresholds are breached.

  • Create a new file called SecurityAlert.java in src/main/java/com/nextwork/audit/model/ by pasting the following code:
package com.nextwork.audit.model;

import java.time.Instant;

// Represents a detected credential-stuffing attack from a single source IP
public record SecurityAlert(
    String sourceIp,
    int failedCount,
    int uniqueAccountCount,
    Instant windowStart,
    Instant windowEnd
) {}

What does this record hold?

SecurityAlert captures all the context needed to investigate a detected attack: which IP triggered it, how many failed attempts occurred, how many unique accounts were targeted, and the exact time window boundaries. This gets produced to the security-alerts topic for downstream consumption by Logstash.

  • Save SecurityAlert.java.
  • Confirm both files appear in the model package alongside AuditEvent.java and EventType.java.

File not in the right package?

Make sure both files are inside src/main/java/com/nextwork/audit/model/. If you accidentally placed them in a different folder, the package declaration will not match the directory structure and compilation will fail.

Help me fix my Java package structure for AttackAggregator and SecurityAlert.

Build the Kafka Streams topology with windowed aggregation

With your state model ready, you can now wire up the Kafka Streams topology. Spring Boot auto-configures the required KafkaStreamsConfiguration bean when kafka-streams is on the classpath and @EnableKafkaStreams is present. You just need to define a @Bean method that builds the processing graph.

  • Create a new package folder src/main/java/com/nextwork/audit/streams/.
  • Inside it, create a file called SecurityStreamsTopology.java and paste the following code:
package com.nextwork.audit.streams;

import com.nextwork.audit.model.AttackAggregator;
import com.nextwork.audit.model.AuditEvent;
import com.nextwork.audit.model.EventType;
import com.nextwork.audit.model.SecurityAlert;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.support.serializer.JsonSerde;

import java.time.Duration;
import java.time.Instant;

@Configuration
@EnableKafkaStreams
public class SecurityStreamsTopology {
    private static final Logger log = LoggerFactory.getLogger(SecurityStreamsTopology.class);

    @Bean
    public KStream<String, AuditEvent> auditStream(StreamsBuilder builder) {
        // Define custom Serdes for JSON serialization of each model type
        JsonSerde<AuditEvent> auditSerde = new JsonSerde<>(AuditEvent.class);
        JsonSerde<AttackAggregator> aggregatorSerde = new JsonSerde<>(AttackAggregator.class);
        JsonSerde<SecurityAlert> alertSerde = new JsonSerde<>(SecurityAlert.class);

        // Consume from the audit-events topic, keyed by sourceIp (String)
        KStream<String, AuditEvent> stream = builder.stream(
            "audit-events",
            Consumed.with(Serdes.String(), auditSerde)
        );

        // Filter, window, and aggregate failed login events per IP
        stream
            .filter((ip, event) -> event.eventType() == EventType.LOGIN_FAILED)
            .groupByKey(Grouped.with(Serdes.String(), auditSerde))
            .windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(
                Duration.ofSeconds(60),
                Duration.ofSeconds(10)
            ))
            .aggregate(
                AttackAggregator::new,
                (ip, event, agg) -> agg.increment(event.userId()),
                Materialized.with(Serdes.String(), aggregatorSerde)
            )
            .toStream()
            .filter((windowedKey, agg) ->
                agg.getFailedCount() >= 20 && agg.getUniqueUserIds().size() >= 10
            )
            .map((windowedKey, agg) -> {
                SecurityAlert alert = new SecurityAlert(
                    windowedKey.key(),
                    agg.getFailedCount(),
                    agg.getUniqueUserIds().size(),
                    Instant.ofEpochMilli(windowedKey.window().start()),
                    Instant.ofEpochMilli(windowedKey.window().end())
                );
                log.warn("SECURITY ALERT: Credential stuffing detected from IP {} - {} failed attempts across {} accounts",
                    alert.sourceIp(), alert.failedCount(), alert.uniqueAccountCount());
                return KeyValue.pair(windowedKey.key(), alert);
            })
            .to("security-alerts", Produced.with(Serdes.String(), alertSerde));

        return stream;
    }
}

What does this topology do?

This is a single continuous processing pipeline that reads from audit-events and writes alerts to security-alerts. Here is how each stage works:

  • filter() discards any event that is not a LOGIN_FAILED. Only failed logins matter for brute-force detection.
  • groupByKey() groups events by their existing key (sourceIp). Because the producer already keyed by IP, no repartition is needed.
  • windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(...)) creates overlapping 60-second windows. Every new event creates a fresh window boundary, ensuring no 60-second span is missed.
  • aggregate() initializes an empty AttackAggregator for each window, then calls increment() on every incoming event to update the count and track unique users.
  • The downstream filter() checks the compound threshold: 20+ failures AND 10+ unique accounts.
  • map() transforms matching aggregations into SecurityAlert objects and logs a warning. Then to() produces alerts to the security-alerts topic.

Why SlidingWindows instead of SessionWindows?

SessionWindows close during gaps in activity. If an attacker pauses for a few seconds between bursts, a session window would split the attack into separate sessions and potentially miss the threshold. SlidingWindows creates a new window boundary for every arriving event, guaranteeing that every possible 60-second span is evaluated.

  • Save SecurityStreamsTopology.java.
  • Your project should now compile without errors. You will not see a runtime change yet because the Spring Boot application is not running.

Seeing compilation errors?

Check that your import for SlidingWindows resolves from org.apache.kafka.streams.kstream.*. If your IDE shows it as unresolved, confirm that the kafka-streams dependency is in pom.xml and run a Maven reload.

Also verify that AttackAggregator.java and SecurityAlert.java are in the com.nextwork.audit.model package. The imports in SecurityStreamsTopology.java reference them by full package path.

Help me fix compilation errors in my Kafka Streams topology class.

Understand the threshold detection logic

The compound threshold in this topology is what distinguishes a credential-stuffing attack from a single user mistyping their password repeatedly.

Why two thresholds instead of one?

A single user forgetting their password might generate 5-10 failed attempts from one IP. That would trigger a simple count-based alert. By requiring both 20+ failures AND 10+ distinct accounts, you filter out legitimate forgetfulness and only fire on attacks that spray credentials across many accounts.

This compound logic cannot be done with a simple count() call. It requires aggregate() with a custom state object that tracks both metrics simultaneously. That is exactly what AttackAggregator provides.

What is the Kafka Streams application ID?

The spring.kafka.streams.application-id property you set in application.properties (security-audit-streams) serves as the consumer group ID for the Streams topology. This means the Streams app and Logstash use different group IDs, enabling independent consumption of the same audit-events topic without interfering with each other.

✔️ Awesome, I've got everything!

Great. Double check you have saved all three new files before moving on.

ⓧ I'd like to double check the full code

Here is the complete SecurityStreamsTopology.java file:

package com.nextwork.audit.streams;

import com.nextwork.audit.model.AttackAggregator;
import com.nextwork.audit.model.AuditEvent;
import com.nextwork.audit.model.EventType;
import com.nextwork.audit.model.SecurityAlert;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.support.serializer.JsonSerde;

import java.time.Duration;
import java.time.Instant;

@Configuration
@EnableKafkaStreams
public class SecurityStreamsTopology {
    private static final Logger log = LoggerFactory.getLogger(SecurityStreamsTopology.class);

    @Bean
    public KStream<String, AuditEvent> auditStream(StreamsBuilder builder) {
        JsonSerde<AuditEvent> auditSerde = new JsonSerde<>(AuditEvent.class);
        JsonSerde<AttackAggregator> aggregatorSerde = new JsonSerde<>(AttackAggregator.class);
        JsonSerde<SecurityAlert> alertSerde = new JsonSerde<>(SecurityAlert.class);

        KStream<String, AuditEvent> stream = builder.stream(
            "audit-events",
            Consumed.with(Serdes.String(), auditSerde)
        );

        stream
            .filter((ip, event) -> event.eventType() == EventType.LOGIN_FAILED)
            .groupByKey(Grouped.with(Serdes.String(), auditSerde))
            .windowedBy(SlidingWindows.ofTimeDifferenceAndGrace(
                Duration.ofSeconds(60),
                Duration.ofSeconds(10)
            ))
            .aggregate(
                AttackAggregator::new,
                (ip, event, agg) -> agg.increment(event.userId()),
                Materialized.with(Serdes.String(), aggregatorSerde)
            )
            .toStream()
            .filter((windowedKey, agg) ->
                agg.getFailedCount() >= 20 && agg.getUniqueUserIds().size() >= 10
            )
            .map((windowedKey, agg) -> {
                SecurityAlert alert = new SecurityAlert(
                    windowedKey.key(),
                    agg.getFailedCount(),
                    agg.getUniqueUserIds().size(),
                    Instant.ofEpochMilli(windowedKey.window().start()),
                    Instant.ofEpochMilli(windowedKey.window().end())
                );
                log.warn("SECURITY ALERT: Credential stuffing detected from IP {} - {} failed attempts across {} accounts",
                    alert.sourceIp(), alert.failedCount(), alert.uniqueAccountCount());
                return KeyValue.pair(windowedKey.key(), alert);
            })
            .to("security-alerts", Produced.with(Serdes.String(), alertSerde));

        return stream;
    }
}

Here is the complete AttackAggregator.java:

package com.nextwork.audit.model;

import java.util.HashSet;
import java.util.Set;

public class AttackAggregator {
    private int failedCount;
    private Set<String> uniqueUserIds;

    public AttackAggregator() {
        this.failedCount = 0;
        this.uniqueUserIds = new HashSet<>();
    }

    public AttackAggregator increment(String userId) {
        this.failedCount++;
        this.uniqueUserIds.add(userId);
        return this;
    }

    public int getFailedCount() { return failedCount; }
    public Set<String> getUniqueUserIds() { return uniqueUserIds; }
    public void setFailedCount(int failedCount) { this.failedCount = failedCount; }
    public void setUniqueUserIds(Set<String> uniqueUserIds) { this.uniqueUserIds = uniqueUserIds; }
}

Here is the complete SecurityAlert.java:

package com.nextwork.audit.model;

import java.time.Instant;

public record SecurityAlert(
    String sourceIp,
    int failedCount,
    int uniqueAccountCount,
    Instant windowStart,
    Instant windowEnd
) {}

Your detection topology is wired and ready. Next up, you will configure the ELK pipeline to consume from both topics and simulate a live attack to see the full system in action.

Configure ELK Pipeline and Simulate Attacks

Your Kafka Streams topology can detect credential-stuffing attacks in real time. But detection alone is not enough. Security teams need to search historical audit logs, correlate events, and visualize attack patterns over time.

In this step, you will wire up the ELK stack to consume from your Kafka topics using independent consumer groups. Then you will simulate a brute-force attack and watch the entire pipeline fire end-to-end.

In this step, get ready to:
  • Write a Logstash pipeline configuration that consumes from both Kafka topics.
  • Build an attack simulator that generates a realistic credential-stuffing burst.
  • Verify events and alerts appear in Kibana with a data table visualization.
Write the Logstash pipeline configuration

Logstash acts as the bridge between Kafka and Elasticsearch. It needs two separate Kafka inputs because the audit-events topic and the security-alerts topic should land in different Elasticsearch indices.

Each input uses its own consumer group ID. This means Logstash reads the topics independently from the Kafka Streams topology. Both consumers see every message without interfering with each other.

  • Create the Logstash pipeline config file by running:
touch logstash/pipeline/logstash.conf
  • Confirm the file exists by running:
ls logstash/pipeline/

You should see logstash.conf in the output.

  • Open logstash/pipeline/logstash.conf in your editor and paste the following configuration:
input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["audit-events"]
    group_id => "elk-ingestion"
    codec => json
    decorate_events => true
    tags => ["audit"]
  }
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["security-alerts"]
    group_id => "elk-alerts"
    codec => json
    decorate_events => true
    tags => ["alert"]
  }
}

filter {
  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
  }
}

output {
  if "audit" in [tags] {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      index => "audit-events-%{+YYYY.MM.dd}"
    }
  } else if "alert" in [tags] {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      index => "security-alerts-%{+YYYY.MM.dd}"
    }
  }
}

What does this configuration do?

The input section defines two independent Kafka consumers. Each has its own group_id so Kafka tracks their offsets separately. The tags field labels each event so the output section knows which index to route it to.

The filter section parses the timestamp field from your events into Elasticsearch's @timestamp so time-based queries and dashboards work correctly.

The output section uses conditionals on the tags to route audit events into daily audit-events-YYYY.MM.dd indices and alerts into security-alerts-YYYY.MM.dd indices.

Why two consumer groups instead of one?

Kafka allows multiple consumer groups to read the same topic independently. The Streams topology uses security-audit-streams as its group ID, while Logstash uses elk-ingestion and elk-alerts. Each group maintains its own offset position, so they all receive every message without competing.

  • Save logstash.conf.
  • Restart Logstash so it picks up the new pipeline config by running:
docker compose restart logstash
  • Confirm Logstash is running by checking container status:
docker compose ps

You should see the logstash container with a status of Up.

Logstash container keeps restarting?

Check the Logstash logs for syntax errors in your pipeline config by running docker compose logs logstash. Common issues include missing closing braces or incorrect quoting in the config file.

Make sure the bootstrap_servers value is kafka:9092 (the container hostname), not localhost:9092. Logstash runs inside Docker and needs the container network name to reach Kafka.

Help me debug my Logstash configuration

Build the attack simulator

A real credential-stuffing attack sends many failed login attempts from a single IP address, targeting multiple user accounts in rapid succession. Your simulator needs to replicate this pattern so the Kafka Streams topology triggers an alert.

You will implement a CommandLineRunner that fires automatically when the Spring Boot app starts. It will send 25 failed logins from one IP targeting 12 accounts, interspersed with legitimate logins from other IPs to prove the filter only catches the attacker.

  • Create the file src/main/java/com/nextwork/audit/simulator/AttackSimulator.java and paste the following code:
package com.nextwork.audit.simulator;

import com.nextwork.audit.model.AuditEvent;
import com.nextwork.audit.model.EventType;
import com.nextwork.audit.producer.AuditEventProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

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

@Component
public class AttackSimulator implements CommandLineRunner {
    private static final Logger log = LoggerFactory.getLogger(AttackSimulator.class);
    private final AuditEventProducer producer;

    public AttackSimulator(AuditEventProducer producer) {
        this.producer = producer;
    }

    @Override
    public void run(String... args) throws Exception {
        String attackerIp = "203.0.113.42";
        log.info("Starting attack simulation from IP: {}", attackerIp);

        // Simulate 25 failed logins from attacker IP targeting 12 different accounts
        for (int i = 0; i < 25; i++) {
            String targetUser = "user" + (i % 12);
            AuditEvent failedLogin = new AuditEvent(
                UUID.randomUUID().toString(),
                EventType.LOGIN_FAILED,
                targetUser,
                attackerIp,
                Instant.now(),
                Map.of("attempt", String.valueOf(i + 1))
            );
            producer.send(failedLogin);

            // Intersperse legitimate logins from other IPs
            if (i % 5 == 0) {
                AuditEvent legitimateLogin = new AuditEvent(
                    UUID.randomUUID().toString(),
                    EventType.LOGIN_SUCCESS,
                    "legitimate-user-" + i,
                    "10.0.0." + (i + 1),
                    Instant.now(),
                    Map.of("source", "mobile-app")
                );
                producer.send(legitimateLogin);
            }
            Thread.sleep(50);
        }
        log.info("Attack simulation complete. Watch for SECURITY ALERT in logs.");
    }
}

What does this simulator do?

  • It sends 25 LOGIN_FAILED events from attacker IP 203.0.113.42 targeting 12 different user accounts (i % 12 cycles through user0 to user11).
  • Every 5th iteration, a legitimate LOGIN_SUCCESS event is sent from a different IP. This proves the Streams filter only catches the attacker's IP, not legitimate traffic.
  • The 50ms sleep between events simulates a rapid-fire attack. All 25 events arrive well within the 60-second sliding window, which exceeds the threshold of 20 failures across 10+ accounts.
  • Save AttackSimulator.java.

Unsure where to create the file?

The full path is src/main/java/com/nextwork/audit/simulator/AttackSimulator.java. You need to create the simulator folder inside the audit package directory first.

Help me find where to put my AttackSimulator class

✔️ Awesome, I've got everything!

Great. Make sure you have saved both logstash.conf and AttackSimulator.java before moving on.

ⓧ I'd like to double check the full code

Here is the complete logstash.conf:

input {
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["audit-events"]
    group_id => "elk-ingestion"
    codec => json
    decorate_events => true
    tags => ["audit"]
  }
  kafka {
    bootstrap_servers => "kafka:9092"
    topics => ["security-alerts"]
    group_id => "elk-alerts"
    codec => json
    decorate_events => true
    tags => ["alert"]
  }
}

filter {
  date {
    match => ["timestamp", "ISO8601"]
    target => "@timestamp"
  }
}

output {
  if "audit" in [tags] {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      index => "audit-events-%{+YYYY.MM.dd}"
    }
  } else if "alert" in [tags] {
    elasticsearch {
      hosts => ["http://elasticsearch:9200"]
      index => "security-alerts-%{+YYYY.MM.dd}"
    }
  }
}

And here is the complete AttackSimulator.java:

package com.nextwork.audit.simulator;

import com.nextwork.audit.model.AuditEvent;
import com.nextwork.audit.model.EventType;
import com.nextwork.audit.producer.AuditEventProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

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

@Component
public class AttackSimulator implements CommandLineRunner {
    private static final Logger log = LoggerFactory.getLogger(AttackSimulator.class);
    private final AuditEventProducer producer;

    public AttackSimulator(AuditEventProducer producer) {
        this.producer = producer;
    }

    @Override
    public void run(String... args) throws Exception {
        String attackerIp = "203.0.113.42";
        log.info("Starting attack simulation from IP: {}", attackerIp);

        // Simulate 25 failed logins from attacker IP targeting 12 different accounts
        for (int i = 0; i < 25; i++) {
            String targetUser = "user" + (i % 12);
            AuditEvent failedLogin = new AuditEvent(
                UUID.randomUUID().toString(),
                EventType.LOGIN_FAILED,
                targetUser,
                attackerIp,
                Instant.now(),
                Map.of("attempt", String.valueOf(i + 1))
            );
            producer.send(failedLogin);

            // Intersperse legitimate logins from other IPs
            if (i % 5 == 0) {
                AuditEvent legitimateLogin = new AuditEvent(
                    UUID.randomUUID().toString(),
                    EventType.LOGIN_SUCCESS,
                    "legitimate-user-" + i,
                    "10.0.0." + (i + 1),
                    Instant.now(),
                    Map.of("source", "mobile-app")
                );
                producer.send(legitimateLogin);
            }
            Thread.sleep(50);
        }
        log.info("Attack simulation complete. Watch for SECURITY ALERT in logs.");
    }
}
Run the application and verify the pipeline end-to-end

Everything is connected. When you start the Spring Boot app, the attack simulator fires immediately, events flow into Kafka, the Streams topology detects the attack, and Logstash ships everything to Elasticsearch for visualization in Kibana.

  • Start the Spring Boot application by running:
mvn spring-boot:run
  • Watch the console output. Within a few seconds you should see the attack simulation start, followed by a SECURITY ALERT log line from the Kafka Streams topology.

What should I see in the console?

You will first see Starting attack simulation from IP: 203.0.113.42 from the simulator. After about 1-2 seconds, the Streams topology outputs a WARN-level message containing SECURITY ALERT: Credential stuffing detected with the attacker's IP, failed count, and unique account count.

Now let's verify the data made it through Logstash into Elasticsearch and explore it in Kibana.

  • Wait about 30 seconds for Logstash to process the events, then verify Elasticsearch received documents by running:
curl http://localhost:9200/audit-events-*/_count

You should see a JSON response with a count value greater than 0. This confirms audit events reached Elasticsearch.

  • Check the security alerts index too by running:
curl http://localhost:9200/security-alerts-*/_count

This should also show a count greater than 0.

Seeing a count of 0?

Logstash may need more time to connect to Kafka and consume messages. Check Logstash logs with docker compose logs logstash --tail 50 and look for messages about pipeline startup or Kafka connection errors.

If Logstash just started, wait 60 seconds and retry the curl command. The Kafka consumer needs time to join its group and begin fetching.

Help me debug why my Elasticsearch indices are empty

  • Open Kibana in your browser by navigating to http://localhost:5601.
  • Click the hamburger menu (three horizontal lines) in the top left.
  • Click **Discover** under the **Analytics** section.
  • Click **Create a data view**.
  • Enter audit-events-* in the **Index pattern** field.
  • Select @timestamp as the **Timestamp field**.
  • Click **Save data view to Kibana**.
  • Repeat the process to create a second data view with the pattern security-alerts-* and the same @timestamp field.

You should now see documents listed in the Discover view when you select each data view. The audit events data view shows both failed and successful logins. The security alerts data view shows the triggered alert records.

  • Now build a data table to visualize the alerts. Click the hamburger menu again.
  • Click **Visualize Library** under **Analytics**.
  • Click **Create visualization**.
  • Select **Lens** as the visualization type.
  • In the data view dropdown at the top left, select security-alerts-*.
  • Change the visualization type to **Table** using the chart type selector.
  • Drag the sourceIp field from the left panel into the table.
  • Drag failedCount and uniqueAccountCount into the table as well.
  • Click **Save** in the top right and name your visualization.

Your data table should show at least one row with IP 203.0.113.42, a failed count of 20 or more, and a unique account count of 10 or more.

Don't see any data in the table?

Make sure the time range selector in the top right of Kibana includes the current time. Click it and select **Last 15 minutes** or **Today** to ensure your recent events are in scope.

If the fields don't appear in the left panel, click the refresh button next to the data view dropdown. Kibana caches the field list and may need a manual refresh after new indices are created.

Help me troubleshoot my Kibana visualization

Your full security audit pipeline is now live. Events flow from the producer through Kafka, get analyzed by Streams for threats, and land in Elasticsearch for compliance search and visualization. Next up, you will extend this pipeline with geo-enrichment in the Secret Mission.

Secret mission

Geo-Enrichment and Attack Map

Your pipeline detects attacks, but where are they coming from? In this bonus challenge, you'll enrich security alerts with geographic data by mapping attacker IPs to countries using a Logstash lookup table. Then you'll build a Kibana Maps visualization showing attack origins as proportional circles on a world map.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally using Docker containers, so there are no ongoing cloud costs. However, the containers consume local disk space and RAM while running.

Memory usage

Elasticsearch alone uses approximately 2GB of RAM. If you leave the containers running, you may notice your machine slowing down during other tasks.

Resources you used:

  • Docker containers: Kafka, Elasticsearch, Logstash, and Kibana.
  • Docker volumes: kafka-data and es-data (persistent data storage for Kafka and Elasticsearch).
  • Local project files: the security-audit-pipeline directory on your Desktop.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Keep everything running

No action needed. Choose this if you are still actively experimenting with the pipeline or want to try sending more events through the system.

  • Your Docker containers will continue running in the background.
  • Kibana remains accessible at localhost:5601.
  • Elasticsearch stays accessible at localhost:9200.
  • Keep in mind that Elasticsearch uses significant memory (~2GB) while running.

Pause - I'll come back to this later

Shut down the running containers to free up memory, but keep your data volumes intact so you can pick up where you left off.

  • Stop all containers by running this command from your project directory:
docker compose stop
  • Your Kafka data and Elasticsearch indices are preserved in Docker volumes.
  • When you are ready to resume, run docker compose start from the same directory to bring everything back up.

Delete - I don't want to use this again

Remove all containers, networks, and persistent volumes to start completely fresh.

  • Remove all Docker containers and volumes by running this command from your project directory:
docker compose down -v

This removes the Kafka, Elasticsearch, Logstash, and Kibana containers along with the kafka-data and es-data volumes.

  • Delete the local project files by running:
cd ~/Desktop
rm -rf security-audit-pipeline

On Windows, use this command instead:

cd ~\Desktop
rmdir /s /q security-audit-pipeline
  • After running these commands, all project resources are fully removed from your machine.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade security audit pipeline that streams events through Kafka, detects credential-stuffing attacks in real time with Kafka Streams, and visualizes threats in Kibana.

You've learned how to:

  • Implemented a Kafka Streams topology using SlidingWindows for real-time credential-stuffing detection with compound threshold logic tracking both failed login counts and unique accounts.
  • Orchestrated Apache Kafka 4.3.0 in KRaft mode with the full ELK stack using Docker Compose, understanding how KRaft replaced ZooKeeper's coordination responsibilities.
  • Configured a Logstash pipeline with independent consumer groups demonstrating Kafka's multi-consumer architecture for parallel stream processing.
  • Secret Mission: Added geo-enrichment to security alerts and built a Kibana Maps visualization showing global attack origins as proportional circles.

Ready to quiz yourself?