x-defaults: &defaults
  restart: unless-stopped
  logging:
    driver: json-file
    options:
      max-size: 100m
      max-file: "3"

services:
  jobmanager:
    <<: *defaults
    image: ${GLOBAL_REGISTRY:-}flink:${FLINK_VERSION:-1.20.0-scala_2.12-java11}
    hostname: jobmanager
    command: jobmanager
    environment:
      TZ: ${TZ:-UTC}
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        jobmanager.memory.process.size: 1600m
        taskmanager.memory.process.size: 1600m
        taskmanager.numberOfTaskSlots: 2
    volumes:
      - flink_data:/opt/flink/data
      # Custom Flink jobs
      # - ./jobs:/opt/flink/jobs
    ports:
      - "${FLINK_JOBMANAGER_RPC_PORT_OVERRIDE:-6123}:6123"
      - "${FLINK_JOBMANAGER_UI_PORT_OVERRIDE:-8081}:8081"
    deploy:
      resources:
        limits:
          cpus: ${FLINK_JOBMANAGER_CPU_LIMIT:-2.0}
          memory: ${FLINK_JOBMANAGER_MEMORY_LIMIT:-2G}
        reservations:
          cpus: ${FLINK_JOBMANAGER_CPU_RESERVATION:-0.5}
          memory: ${FLINK_JOBMANAGER_MEMORY_RESERVATION:-1G}
    healthcheck:
      test: ["CMD-SHELL", "wget --no-verbose --tries=1 --spider http://localhost:8081 || exit 1"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 30s

  taskmanager:
    <<: *defaults
    image: ${GLOBAL_REGISTRY:-}flink:${FLINK_VERSION:-1.20.0-scala_2.12-java11}
    command: taskmanager
    environment:
      TZ: ${TZ:-UTC}
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        jobmanager.memory.process.size: 1600m
        taskmanager.memory.process.size: 1600m
        taskmanager.numberOfTaskSlots: 2
    volumes:
      - flink_data:/opt/flink/data
      # Custom Flink jobs
      # - ./jobs:/opt/flink/jobs
    depends_on:
      jobmanager:
        condition: service_healthy
    deploy:
      resources:
        limits:
          cpus: ${FLINK_TASKMANAGER_CPU_LIMIT:-2.0}
          memory: ${FLINK_TASKMANAGER_MEMORY_LIMIT:-2G}
        reservations:
          cpus: ${FLINK_TASKMANAGER_CPU_RESERVATION:-0.5}
          memory: ${FLINK_TASKMANAGER_MEMORY_RESERVATION:-1G}
    healthcheck:
      test: ["CMD-SHELL", "ps aux | grep -v grep | grep -q taskmanager || exit 1"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 30s

volumes:
  flink_data: