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

• schema-path-account-create.yml
• schema-path-account-update.yml
• validations-path.yml

📊 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