Kafka Streams Delayed Message Processor

A Kafka Streams application that implements delayed message processing. Messages sent to the input topic are processed after a configurable delay period (default: 15 minutes).

Features

• Configurable delay duration
• Persistent state storage for delayed messages
• Automatic message forwarding after the delay period
• Configurable topic names
• Error handling and logging
• Unit tests included

Architecture

The application uses:

• State Store: Persistent storage for delayed messages
• Punctuator: Periodic checking for ready-to-process messages
• Kafka Streams: Stream processing framework
• Jackson: JSON serialization/deserialization

Configuration

Configuration is provided in src/main/resources/application.properties:

# Kafka configuration
kafka.bootstrap.servers=localhost:9092
kafka.application.id=kafka-streams-delayed-processor
kafka.client.id=kafka-streams-delayed-client

# Input and output topics
kafka.input.topic=delayed-messages-input
kafka.output.topic=delayed-messages-output

# Delay configuration (in minutes)
delay.duration.minutes=15

# State store configuration
state.store.name=delayed-messages-store

Building the Application

Prerequisites

• Java 11 or higher
• Apache Kafka (running locally or remotely)
• Maven 3.6+

Running Locally

mvn clean compile exec:java -Dexec.mainClass="com.example.kafka.streams.DelayedMessageApplication"

Building

mvn clean package

Containerization

# Build an image
docker image build -t selcukusta/kafka_stream_delay_process:1.0.0 .

# Start a new container (.env file is not mandatory. If it's not defined, default configuration will be bound)
docker container run --env-file .env selcukusta/kafka_stream_delay_process:1.0.0

Running the Application

# Start the Kafka Streams application
java -jar target/kafka-streams-delayed-processor-1.0.0.jar

# Start with custom configurations from environment variables
KAFKA_BOOTSTRAP_SERVERS=[YOUR_BOOTSTRAP_URLS] DELAY_DURATION_MINUTES=[YOUR_CUSTOM_DURATION] java -jar target/kafka-streams-delayed-processor-1.0.0.jar

# Or using Maven
mvn exec:java -Dexec.mainClass="com.example.kafka.streams.DelayedMessageApplication"

Usage

1. Create Topics (if they don't exist)

kafka-topics.sh --create --topic delayed-messages-input --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1
kafka-topics.sh --create --topic delayed-messages-output --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1

2. Consume Processed Messages

kafka-console-consumer.sh --topic delayed-messages-output --bootstrap-server localhost:9092 --from-beginning

Message Format

Messages must be in JSON format with the following structure:

Key: HBV000010KCBY

Value:

{
  "name": "Fizello Dunkin Donuts - Drunken Grownups Kupa Bardak",
  "brand": "Fizello",
  "barcode": ["FIZELLO-MUG0332513"],
  "qualityScore": 74.4,
  "qualityStatus": "GOOD"
}

Testing

Run unit tests:

mvn test

Monitoring

The application includes SLF4J logging for:

• Message processing events
• State store operations
• Error handling
• Application lifecycle events

Scaling

The application can be scaled by:

• Increasing num.stream.threads in the configuration
• Running multiple instances (Kafka will partition the load)
• Adjusting cache settings for better performance