kafka-and-postgres.yml

version: "2"

services:
  kafka:
    image: docker.io/bitnami/kafka:3.5
    ports:
      - "9092:9092"
    volumes:
      - "kafka_data:/bitnami"
    environment:
      # KRaft settings
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      # Listeners
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
      # Other
      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true

  db:
    image: postgres:16
    restart: always
    environment:
      - POSTGRES_DB=mydb
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
    ports:
      - '5432:5432'
    volumes: 
      - db:/var/lib/postgresql/data

volumes:
  kafka_data:
    driver: local
  db:
    driver: local

users.json

{
    "generators": [
        {
            "connection": "dev-kafka",
            "topic": "users",
            "value": {
                "name": {
                    "_gen": "string",
                    "expr": "#{Name.firstName}"
                },
                "age": {
                    "_gen": "integer",
                    "n": {
                        "_gen": "uniformDistribution",
                        "bounds": [18, 80]
                    }
                },
                "level": {
                    "_gen": "weightedOneOf",
                    "choices": [
                        { "weight": 1, "value": "gold" },
                        { "weight": 4, "value": "silver" },
                        { "weight": 5, "value": "bronze" }
                    ]
                }
            }
        }
    ],
    "connections": {
        "dev-kafka": {
            "kind": "kafka",
            "producerConfigs": {
                "bootstrap.servers": "localhost:9092",
                "key.serializer": "io.shadowtraffic.kafka.serdes.JsonSerializer",
                "value.serializer": "io.shadowtraffic.kafka.serdes.JsonSerializer"
            }
        },
        "pg": {
            "kind": "postgres",
            "connectionConfigs": {
                "host": "localhost",
                "port": 5432,
                "username": "postgres",
                "password": "postgres",
                "db": "mydb"
            }
        }
    }
}