Kafka Streams Processor

Java-based Kafka Streams application for real-time data processing with deduplication, validation, and state store management.

📁 Project Structure

Kafka-Streams-Processor/
├── 📂 src/main/java/
│   ├── 📂 consumers/                    # Kafka consumer implementations
│   └── 📂 org/example/kafka/
│       ├── 📂 config/                   # Kafka configuration
│       ├── 📂 processor/                # Stream processors
│       ├── 📂 streaming/                # Main streaming applications
│       ├── 📂 topology/                 # Stream topology definitions
│       └── 📂 validator/                # Message validation
├── 📂 src/main/resources/               # Configuration files
├── 📄 docker-compose.yml               # Kafka cluster setup
├── 📄 pom.xml                          # Maven dependencies
└── 📄 runbook.txt                      # Execution instructions

🚀 Quick Start

1. Start Kafka Cluster

docker-compose up -d

2. Build Application

mvn clean compile

3. Run Stream Processor

mvn exec:java -Dexec.mainClass="org.example.kafka.streaming.KafkaStreamProcessor"

4. Send Test Messages

# Connect to Kafka container
docker exec -it kafka-streams-processor-kafka-1 bash

# Start producer
kafka-console-producer --broker-list localhost:19092 --topic input-topic-account-create \
  --property "parse.key=true" --property "key.separator=:"

🔧 Components

Stream Processors

Class	Purpose
`KafkaStreamProcessor`	Main streaming application entry point
`KafkaJoinStreamingProcessor`	Stream joining operations
`StreamProcessor`	Core stream processing logic
`UniqueRecordProcessor`	Deduplication processing

Configuration

File	Purpose
`KafkaConfig.java`	Kafka streams configuration
`application.yaml`	Application properties
`config.yaml`	Custom configuration

Topics

input-topic-account-create - Account creation events
input-topic-account-update - Account update events
Output topics configured in topology

🛠️ Features

Deduplication: Unique record processing with state stores
Validation: Schema-based message validation
Join Operations: Stream-to-stream and stream-to-table joins
State Management: RocksDB-backed state stores
Error Handling: Dead letter queue patterns
Monitoring: Application metrics and logging

📋 Prerequisites

Java 11+
Maven 3.6+
Docker and Docker Compose
Apache Kafka 2.7+

🔧 Configuration

Environment Variables

KAFKA_BOOTSTRAP_SERVERS=localhost:19092
KAFKA_APPLICATION_ID=KafkaStreamProcessor

State Stores

state-store-account-create - Account creation deduplication
state-store-account-update - Account update deduplication

Schema Files

📊 Sample Data

Account creation event:

{
  "event_id": "unique-event-id-001",
  "timestamp": "2024-07-25T12:00:00Z",
  "account_created": {
    "account": {
      "id": "acc-001",
      "name": "Account Name",
      "status": "active",
      "permitted_denominations": ["GBP", "USD"]
    }
  }
}

🔍 Monitoring

Application Logs

tail -f logs/application.log

Kafka Topics

# List topics
kafka-topics --bootstrap-server localhost:19092 --list

# Monitor consumer group
kafka-consumer-groups --bootstrap-server localhost:19092 --describe --group KafkaStreamProcessor

State Store Inspection

# Check RocksDB state stores
ls -la KafkaStreamProcessor/

🏗️ Architecture

Input Topics → Stream Processor → Validation → Deduplication → Output Topics
                      ↓
                 State Stores (RocksDB)

🧪 Testing

Manual Testing

See runbook.txt for detailed test scenarios and sample messages.

Integration Tests

mvn test

🛑 Cleanup

# Stop application
Ctrl+C

# Stop Kafka cluster
docker-compose down

# Remove volumes
docker-compose down -v

Real-time data processing with Apache Kafka Streams

FilesExpand file tree

README.md

Latest commit

History