Distributed Job Queue

A distributed asynchronous job processing system built with Spring Boot 4.1, Java 25, Apache Kafka 4.0 (KRaft mode), and PostgreSQL 17. Jobs flow through a lifecycle of QUEUED → PROCESSING → COMPLETED / DEAD_LETTER with exponential backoff retry and worker-lock recovery.

Architecture

                        ┌──────────────────────┐
                        │    Kafka 4.0 (KRaft) │
                        │  topic: job_queue    │
                        └──────────┬───────────┘
                                   │ consume
                         ┌─────────▼─────────┐
                         │   Worker Service  │
                         │ (consumer group:  │
                         │   job-workers)    │
                         │                   │
                         │  ┌─ JobConsumer   │
                         │  ├─ RetryScheduler│
                         │  └─ LockReaper    │
                         └─────────┬─────────┘
                                   │ claim / complete / fail
                         ┌─────────▼─────────┐
                         │    PostgreSQL 17  │
                         │    ┌───────────┐  │
                         │    │   job     │  │
                         │    └───────────┘  │
                         └───────────────────┘

Modules

Module	Purpose
common	Shared DTOs (JobEvent), enums (JobStatus), constants (KafkaTopics)
api	REST API — job creation & query, Kafka producer, exception handling
worker	Kafka consumer, job processing, retry scheduler, lock recovery

Tech Stack

Component	Technology
Runtime	Java 25, Spring Boot 3.5.6
Database	PostgreSQL 17, Hibernate 6.x
Message Queue	Apache Kafka 4.2.0 (KRaft, no ZooKeeper)
API Docs	SpringDoc OpenAPI 2.8.14 (Swagger UI)
Build	Maven 3.9+ (multi-module)
Containerization	Docker + Docker Compose

Job Lifecycle

QUEUED ──► PROCESSING ──► COMPLETED
              │
              ▼ (on failure)
         retry_count < max_retries
              │
              ▼
           QUEUED (after backoff delay)
              │
         retry_count >= max_retries
              │
              ▼
         DEAD_LETTER

• Max retries: 3 (configurable per job)
• Backoff: exponential — 1s, 2s, 4s
• Lock timeout: 5 minutes — stuck PROCESSING jobs are recovered by LockReaper

Quick Start

Prerequisites

• Docker & Docker Compose

Run

cd distributed-job-queue

# Start database and Kafka first
docker compose up -d postgres kafka

# Wait for PostgreSQL to be ready, then create the table
docker compose exec -T postgres psql -U postgres -d jobqueue < init.sql

# Start all services
docker compose up --build

This starts:

• PostgreSQL 17 on port 5434 (host) mapped to 5432 (container)
• Kafka 4.0 (KRaft) on port 9092
• Kafka UI on port 8081
• API Service on port 8080 (Swagger UI: http://localhost:8080/swagger-ui/index.html)
• Worker Service (ephemeral port)

API Endpoints

Method	Endpoint	Description	Response
POST	/api/jobs	Submit a new job	201 Created with job data
GET	/api/jobs?page=&size=	List jobs	200 OK with paginated data
GET	/api/jobs/{id}	Get job status	200 OK or 404 Not Found

# Create a job
curl -X POST http://localhost:8080/api/jobs \
  -H "Content-Type: application/json" \
  -d '{"jobType": "email", "payload": {"to": "user@example.com", "subject": "Hello"}}'

# Get job status
curl http://localhost:8080/api/jobs/{jobId}

Response:

{
  "status": 200,
  "message": "Success",
  "data": {
    "id": "550e8400-e29b-41d4-a716-446655440000",
    "jobType": "email",
    "status": "QUEUED",
    "retryCount": 0,
    "maxRetries": 3,
    "payload": { "to": "user@example.com", "subject": "Hello" },
    "createdAt": "2026-07-09T12:00:00Z"
  }
}

Configuration

All configuration is externalized via environment variables. Copy .env.example to .env in each module and adjust.

Project Structure

distributed-job-queue/
├── pom.xml                          # Parent POM (multi-module)
├── docker-compose.yml               # PostgreSQL, Kafka, Kafka UI, services
├── README.md
│
├── common/
│   ├── pom.xml
│   └── src/main/java/com/jobqueue/common/
│       ├── constants/KafkaTopics.java
│       ├── dto/JobEvent.java
│       └── enums/JobStatus.java
│
├── api/
│   ├── Dockerfile
│   ├── pom.xml
│   ├── .env / .env.example
│       └── src/main/
│           ├── java/com/jobqueue/api/
│           │   ├── ApiApplication.java
│           │   ├── config/KafkaConfig.java
│           │   ├── controller/
│           │   │   ├── GlobalExceptionHandler.java
│           │   │   └── JobController.java
│           │   ├── dto/
│           │   │   ├── JobRequest.java
│           │   │   ├── JobResponse.java
│           │   │   ├── PagedResponse.java
│           │   │   └── Response.java
│           │   ├── exception/
│           │   │   ├── AppException.java
│           │   │   └── NotFoundException.java
│           │   ├── kafka/JobProducer.java
│           │   ├── model/Job.java
│           │   ├── repository/JobRepository.java
│           │   └── service/JobService.java
│           └── resources/
│               └── application.yml
│
└── worker/
    ├── Dockerfile
    ├── pom.xml
    ├── .env / .env.example
    └── src/main/
        ├── java/com/jobqueue/worker/
        │   ├── WorkerApplication.java
        │   ├── config/WorkerInstance.java
        │   ├── consumer/JobConsumer.java
        │   ├── model/Job.java
        │   ├── repository/JobRepository.java
        │   ├── scheduler/
        │   │   ├── LockReaper.java
        │   │   └── RetryScheduler.java
        │   └── service/JobService.java
        └── resources/application.yml

How It Works

Job Submission

POST /api/jobs accepts a JSON payload. JobController delegates to JobService.createJob(), which persists a job in PostgreSQL with status QUEUED and publishes a JobEvent to the job_queue Kafka topic. Returns 201 Created with the job data.

GET /api/jobs/{id} retrieves a job by UUID. Returns 200 OK with the job data or throws NotFoundException (→ 404).

Job Processing

1. JobConsumer listens on job_queue topic
2. On receiving an event, it calls JobService.claimJob() which uses SELECT ... FOR UPDATE SKIP LOCKED to atomically claim the job
3. The worker sets status to PROCESSING, locks the job with its unique worker ID
4. Processing is simulated (2s sleep)
5. On success: status → COMPLETED
6. On failure: failJob() increments retry count, applies exponential backoff, re-queues

Retry Scheduler

RetryScheduler runs every 1 second, queries jobs where status = QUEUED and next_retry_at <= now(), clears the retry time, and re-dispatches them to Kafka.

Lock Reaper

LockReaper runs every 30 seconds, finds jobs stuck in PROCESSING for more than 5 minutes (worker likely crashed), releases the lock, and either re-queues or moves to DEAD_LETTER if max retries are exceeded.

Error Responses

All errors return a consistent JSON envelope:

{
  "status": 404,
  "message": "The requested resource was not found.",
  "data": "Job not found with id: 550e8400-e29b-41d4-a716-446655440000"
}

Database

Schema is managed manually. Before starting the application, run the SQL script at the project root:

# Using Docker (PostgreSQL container must be running)
docker compose exec -T postgres psql -U postgres -d jobqueue < init.sql

This creates the job table. Hibernate runs with ddl-auto: validate to confirm the schema matches the entity on startup.

Column details:

Column	Type	Notes
id	UUID	Auto-generated
job_type	VARCHAR(50)	Required
status	VARCHAR(20)	Default: QUEUED
retry_count	INT	Default: 0
max_retries	INT	Default: 3
payload	JSONB	Default: {}
result	JSONB	Nullable
failure_reason	TEXT	Nullable
locked_by	VARCHAR(100)	Nullable
locked_at	TIMESTAMP	Nullable
next_retry_at	TIMESTAMP	Nullable
created_at	TIMESTAMP	Auto-set
updated_at	TIMESTAMP	Auto-updated

Fault Tolerance

Failure	Handling
Worker crash during processing	LockReaper recovers job after 5min timeout
Kafka broker restart	Messages persisted to disk; clients reconnect
Job processing exception	Retry with exponential backoff (1s → 2s → 4s)
Max retries exhausted	Job moved to DEAD_LETTER status
Concurrent worker contention	SELECT FOR UPDATE SKIP LOCKED prevents double-claiming
Duplicate Kafka messages	Idempotent producer (enable.idempotence: true)