Skip to content
BranchesTags

Repository files navigation

Kafka Queue SDK Example - Docker Setup

This is a complete Docker-based setup for a Kafka-based microservices architecture using kafka_server_sdk==1.0.0 and kafka_queue_client==1.0.0.

The Docker build installs both distributions from the sibling ../kafkaQueueSdk checkout and fails if either imported version is not exactly 1.0.0.

Architecture Overview

Infrastructure Services

  • Kafka (port 9092) - Message broker
  • Kafka UI (port 8020) - Web interface for Kafka management
  • PostgreSQL (port 5433) - Database for services
  • db-migrate - One-shot Alembic upgrade; must complete before DB-backed services start
  • MinIO (ports 9090, 9091) - S3-compatible object storage

Application Services

  • in_gateway (port 7091) - Input gateway service
  • out_gateway (port 7092) - Output gateway service
  • file_storage (port 7093) - File storage service
  • admin (port 7094) - Admin/monitoring service
  • logic (ports 7095, 7096) - Logic processing service
  • hold_service (ports 7097, 7098) - Message hold and management service
  • hold_service_2 (ports 7099, 7100) - Second message hold and management service with custom prefix

Prerequisites

  • Docker
  • Docker Compose v3.8+

Quick Start

Important: The docker-compose.yaml uses a build context that includes the parent kafkaQueueSdk directory. Make sure you run docker-compose from the kafka_queue_sdk_example directory.

  1. Configure the required SDK 1.0 authentication secret:

    Keep the real value out of Git. The same variable is injected into every HTTP service and into queue-admin for its authenticated downstream calls.

    export KAFKA_QUEUE_AUTH_PASSWORD="$(openssl rand -base64 32)"

    .env.example documents all required variable names. Existing local .env files can be retained for the infrastructure values, but any credentials that were ever committed must be rotated.

  2. Start all services:

    cd kafka_queue_sdk_example
docker-compose up -d --build

    Compose runs the idempotent db-migrate job first. It creates an empty schema or upgrades/adopts a supported existing schema without clearing rows.

  3. View logs:

    # All services
docker-compose logs -f

# Specific service
docker-compose logs -f in_gateway

  4. Stop all services:

    docker-compose down

  5. Stop and remove volumes:

    docker-compose down -v

Service Details

in_gateway

  • Purpose: Entry point for incoming requests
  • Port: 7091
  • Dependencies: Kafka, PostgreSQL, S3
  • Config: in_gateway/config.json

out_gateway

  • Purpose: Exit point for outgoing responses
  • Port: 7092
  • Dependencies: Kafka, PostgreSQL
  • Config: out_gateway/config.json

file_storage

  • Purpose: File storage and retrieval service
  • Port: 7093
  • Dependencies: Kafka, PostgreSQL, S3
  • Config: file_storage/config.json

logic

  • Purpose: Business logic processing service
  • Ports: 7095 (main), 7096 (admin)
  • Dependencies: Kafka
  • Config: logic/server_config.json, logic/admin_config.json

hold_service

  • Purpose: Message holding and management service for controlled message flow
  • Ports: 7097 (main), 7098 (admin)
  • Dependencies: Kafka
  • Config: hold_service/config.json, hold_service/server_config.json, hold_service/admin_config.json
  • Features:
    • Holds messages from logic service
    • Provides API for message management (hold/release)
    • Forwards released messages to file_storage
    • SQLite database for message persistence
  • Admin Panel: http://localhost:7097/hold_admin_panel/ (default prefix)

hold_service_2

  • Purpose: Second message holding and management service demonstrating configurable prefix feature
  • Ports: 7099 (main), 7100 (admin)
  • Dependencies: Kafka
  • Config: hold_service_2/config.json, hold_service_2/server_config.json, hold_service_2/admin_config.json
  • Features:
    • Same functionality as hold_service
    • Uses custom prefix hold_admin_panel_2 (configured in config.json)
    • Separate consumer group and database
    • Demonstrates multiple instances with different URL prefixes
  • Admin Panel: http://localhost:7099/hold_admin_panel_2/ (custom prefix)

admin

  • Purpose: Queue administration and monitoring
  • Port: 7094
  • Dependencies: All application services
  • Config: admin/config.json

Kafka Topics

The following Kafka topics are automatically created:

  • logic_in - Input topic for logic service
  • hold_in - Input topic for hold service (from logic)
  • files_proxy - Input topic for file storage (from hold service)
  • events - Events topic for all services
  • out - Output topic

Message Flow

The architecture follows this message flow:

  1. in_gateway receives requests → publishes to logic_in
  2. logic processes messages from logic_in → publishes to hold_in
  3. hold_service receives messages from hold_in → holds/manages them → publishes to files_proxy
  4. file_storage processes messages from files_proxy → publishes to out
  5. out_gateway sends responses from out topic

Access Points

Configuration

Environment Variables

Edit .env file to configure:

  • PostgreSQL credentials
  • S3 access keys
  • KAFKA_QUEUE_AUTH_PASSWORD is deliberately not stored in the tracked JSON configs; export it before every Compose command or provide it through your deployment secret manager.

SDK 1.0 migration

  • Stop every application service before upgrading an existing test queue.

  • Old Kafka/cache messages and old aggregator snapshots are not compatible and may be removed as described in ../kafkaQueueSdk/MIGRATION.md.

  • Build the updated SDK image, start PostgreSQL, and run the same Alembic job used by normal Compose startup:

    ./migrate_sdk_1_db.sh

  • The command is safe for both an empty database and a supported existing database, and can be rerun. It aborts if existing sessions.external_id values contain nulls or duplicates; correct those rows explicitly and rerun. The expected head is 0002_reliability_jobs; it adds the durable outbox and storage-deletion reconciliation schema without deleting existing rows.

  • in_gateway, out_gateway, and file_storage set KAFKA_QUEUE_SCHEMA_MODE=validate and will not start until db-migrate reaches the packaged Alembic head. Keep this setting when adapting the Compose pattern. If applications should migrate on startup instead, remove validate; the SDK default is upgrade, and their DB role then needs DDL privileges.

  • The in-gateway and file-storage limit uploads to 100 MiB. Change max_upload_bytes in their configs and at the reverse proxy together.

  • In-gateway starts its outbox publisher automatically. File storage starts its deletion reconciler automatically. Their explicit outbox_* and deletion_* values in the example are optional retry/lease tuning, not new required infrastructure or secrets. A durable operation may return 202 while retry is pending; monitor old pending jobs and retry counts.

  • All application consumers use auto.offset.reset=earliest, and Compose waits for kafka-init to finish before starting them. Preserve both properties: a broker healthcheck alone does not prove topics exist, and a new group using largest/latest can skip messages sent before its first assignment.

  • Service-admin capture of payloads and headers is explicitly disabled. If you enable it, set retention/quota controls and treat the observer files as sensitive data.

Service Configurations

Each service has its own configuration files in its directory:

  • in_gateway/config.json
  • out_gateway/config.json
  • file_storage/config.json, server_config.json, admin_config.json
  • logic/server_config.json, admin_config.json
  • hold_service/config.json, server_config.json, admin_config.json
  • hold_service_2/config.json, server_config.json, admin_config.json
  • admin/config.json

Health Checks

All infrastructure services have health checks configured:

  • Kafka (KRaft): Broker API readiness
  • PostgreSQL: pg_isready check
  • S3 Server: HTTP endpoint check

Application services wait for healthy infrastructure, successful topic creation, and (where applicable) the Alembic migration/bucket-provisioning jobs before starting.

Logs

Service logs are mounted to local directories:

  • in_gateway/logs/
  • out_gateway/logs/
  • file_storage/logs/
  • logic/logs/
  • hold_service/logs/
  • hold_service_2/logs/
  • admin/logs/

Troubleshooting

Services not starting

# Check service status
docker-compose ps

# Check specific service logs
docker-compose logs <service_name>

Kafka connection issues

# Verify Kafka is healthy
docker-compose ps kafka

# Check Kafka logs
docker-compose logs kafka

Database connection issues

# Verify PostgreSQL is healthy
docker-compose ps postgres_db

# Connect to database
docker-compose exec postgres_db psql -U postgres

Reset everything

# Stop and remove all containers, networks, and volumes
docker-compose down -v

# Rebuild and start
docker-compose up -d --build

Development

Validate the migration contract

Run the config tests with both sibling source packages on PYTHONPATH so that an older installed client cannot shadow the 1.0 checkout:

PYTHONPATH=../kafkaQueueSdk/client:../kafkaQueueSdk \
  ../kafkaQueueSdk/.venv/bin/python -m pytest -q -p no:cacheprovider \
  tests/test_sdk_1_migration.py

After the stack starts, verify public/protected endpoints with:

./test_auth.sh

For the terminal-flow smoke test, submit a new ZIP to /in/send, wait until its messages appear in the primary hold service, unhold them, and then verify both conditions:

  • /files/list_files/by_ext_id/<external-id> contains every submitted entry;
  • /out/session_info/by_external_uid?external_uid=<external-id> reports the session and every entries_info[] item as completed, with file_storage as the terminal step.

Run this check with a new external-id; pre-fix test sessions can legitimately remain failed or incomplete and are not replayed after their offsets were committed.

The reproducible client-driven variant uses the public high-level client and strictly verifies submission IDs, session and entry fields, all 17 stored objects, the combined download, and every per-entry download:

KAFKA_QUEUE_AUTH_PASSWORD='<local-test-password>' \
PYTHONPATH=../kafkaQueueSdk/client:../kafkaQueueSdk \
  ../kafkaQueueSdk/.venv/bin/python tests/live_client_cycle.py

This Compose topology exposes the three client APIs on separate ports, so the script configures incoming_url, progress_url, and files_url. Deployments with a reverse proxy serving all three paths may continue to pass one base_url. Override the smoke-test endpoints with KAFKA_QUEUE_IN_URL, KAFKA_QUEUE_OUT_URL, KAFKA_QUEUE_FILES_URL, and KAFKA_QUEUE_HOLD_URL.

After a successful client cycle, verify the live deletion reconciliation path using that run's external ID:

KAFKA_QUEUE_AUTH_PASSWORD='<local-test-password>' \
KAFKA_QUEUE_TEST_EXTERNAL_ID='<client-live-external-id>' \
PYTHONPATH=../kafkaQueueSdk/client:../kafkaQueueSdk \
  ../kafkaQueueSdk/.venv/bin/python tests/live_deletion_saga.py

The script accepts either immediate 204 or queued 202, then waits until the deleted entry disappears from the public file list. The corresponding storage_deletion_jobs row should end in completed.

Rebuild specific service

docker-compose up -d --build <service_name>

Access service shell

docker-compose exec <service_name> /bin/bash

View service configuration

docker-compose exec <service_name> cat config.json

Network

All services are connected via the app_net bridge network, allowing them to communicate using service names as hostnames.

Volumes

  • postgres_data: PostgreSQL data persistence
  • scalityS3_data: S3 storage data persistence

Framework Documentation

For more details about the kafkaQueueSdk framework, see:

  • ../kafkaQueueSdk/doc/server/quickstart.md
  • ../kafkaQueueSdk/doc/server/configs.md
  • ../kafkaQueueSdk/doc/server/middleware.md

About

Example application for integrating services through the Kafka Queue SDK.

Topics

kafkamessage-queuepython

Resources

Readme
Activity

Stars

0 stars

Watchers

0 watching

Forks

0 forks
Report repository

Releases

Packages

Contributors

Languages