Kafka-Stream-Engine

A production-grade, highly scalable implementation of a real-time event transformation and streaming engine featuring robust Finite State Machine (FSM) validation, multi-version Pydantic schema mapping, Prometheus observability, and database persistence.

This project demonstrates the technical depth of enterprise streaming pipelines, designed clean-room style to show advanced architectural patterns without exposing proprietary domain logic.

---

🌟 Architectural Features

1. Multi-Version Pydantic Schema Mapping (app/models/schemas.py)

   • Utilizes strict Pydantic V2 models for type safety, validation constraints, and serialization.
   • Declares incoming source models (OrderMessageV3) with nested validations and fields mapped via camelCase aliases, alongside subsequent target models representing specific lifecycle events (OrderPendingV3, OrderAcceptedV3, OrderRejectedV3, etc.).

2. Advanced Finite State Machine Validation (app/core/fsm.py)

   • Integrates the python-statemachine library to validate event transitions at run-time.
   • Enforces a formal order lifecycle: OrderPending -> OrderAccepted/OrderRejected -> OrderSettled / OrderRefunded / OrderCancelled.
   • Handles idempotency (timestamp updates) and guards against out-of-order execution by raising explicit lifecycle violations.

3. Persistent State Storage (app/core/db_handler.py & app/models/db.py)

   • Concrete DBHandler subclass extending the Persistence interface.
   • Syncs FSM state directly to the database via SQLAlchemy.
   • Cross-compatible with PostgreSQL (using on_conflict_do_update upserts) and SQLite (for easy local showcase development).

4. Production-Grade Logger & Cycles (app/core/logger.py)

   • Standardized CommonLogger class matching the interface and telemetry output of industrial streaming pipelines.
   • Automatically tracks transaction boundaries via cycle start/end logs.
   • Formats log outputs as structured JSON containing timestamp, correlation-id, local_time, and execution time (in seconds).

5. Hydration-Mode Runner (app/services/hydration.py)

   • Implements a secondary finite processing loop designed to run in batch environments (such as Kubernetes pods scheduled by Airflow).
   • Polls the hydration queue/topic with custom timeouts, selectively processes historical messages tagged with a hydration=true header, and exits cleanly after an idle period.

6. Prometheus Metrics & Observability

   • Collects throughput, processing status, and error counts categorized by exception reasons using prometheus_client.
   • Exposes standard Prometheus metrics via the /metrics endpoint.

---

🛠️ Project Structure

Kafka-Stream-Engine/
│
├── main.py                     # FastAPI web server & ingestion endpoints
├── requirements.txt            # Project dependencies
├── pyproject.toml              # Ruff and styling configurations
├── mypy.ini                    # Mypy type-checking rules
│
├── app/
│   ├── core/
│   │   ├── config.py           # Configuration settings and env parameters
│   │   ├── db_handler.py       # SQLAlchemy database state persistence
│   │   ├── fsm.py              # Order state machine rules & transitions
│   │   ├── interfaces.py       # Interfaces for processing, streams, and DB
│   │   └── logger.py           # Custom cycle-based structured JSON logger
│   │
│   ├── models/
│   │   ├── db.py               # Database schemas and SQLAlchemy helpers
│   │   ├── events.py           # API ingestion envelopes
│   │   └── schemas.py          # V3 Pydantic schemas (Source & Target states)
│   │
│   └── services/
│       ├── engine.py           # High-throughput validation orchestration
│       ├── hydration.py        # Simulated hydration-mode execution loop
│       └── message_processor.py# Version/Type validation and mapping functions
│
├── scripts/
│   ├── consume_stream.py       # Client subscribing to SSE output stream
│   └── simulate_load.py        # Multi-threaded concurrent order ingestion simulator
│
└── tests/
    └── test_engine.py          # Pytest validation suites (ingest, FSM, hydration)

---

🚦 Getting Started

Local Setup

Create a virtual environment and install all dependencies:

make setup

Or manually:

python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

Run Checks & Tests

Execute Ruff lint checks, formatter validation, and Pytest suites:

# Formats code and verifies style rules
venv/bin/ruff format .
venv/bin/ruff check .

# Runs all test suites
venv/bin/pytest tests

---

🚀 Execution Modes

1. Ingestion and Streaming Mode (Real-Time Pipeline)

Start the FastAPI server:

venv/bin/uvicorn main:app --reload --port 8000

In secondary terminals, run the load simulator and subscriber stream client:

# Subscribes to Server-Sent Events (SSE) from the processing engine
python scripts/consume_stream.py

# Generates and ingests mock orders of various states
python scripts/simulate_load.py

2. Hydration Mode (Batch Historical Hydration)

Hydration mode can be triggered and evaluated using the custom showcase endpoint. Send a POST request to trigger the hydration runner, which will consume historical mock events from a simulated queue, filter hydration headers, run validation, and output the processed stream results:

curl -X POST http://localhost:8000/trigger-hydration

---

📈 Monitoring & Health

Once running, monitor metrics and endpoint state:

• API Health: GET http://localhost:8000/health
• Prometheus Metrics: GET http://localhost:8000/metrics