docker-compose.yaml

version: "3.5"

services:
  # Kafka broker on the first node
  kafka1:
    image: confluentinc/cp-kafka:latest
    hostname: kafka1
    container_name: kafka1
    restart: always
    ports:
      - "59092:59092"  # External listener for client connections
      - "29093:29093"  # Controller listener
      - "19092:19092"  # Internal broker-to-broker listener
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,BROKER:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT'
      KAFKA_ADVERTISED_LISTENERS: 'BROKER://${KAFKA_BROKER1_IP}:19092,EXTERNAL://${KAFKA_BROKER1_IP}:59092'
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 2
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@${KAFKA_BROKER1_IP}:29093,2@${KAFKA_BROKER2_IP}:29094'
      KAFKA_LISTENERS: 'BROKER://0.0.0.0:19092,CONTROLLER://0.0.0.0:29093,EXTERNAL://0.0.0.0:59092'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'BROKER'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
      CLUSTER_ID: '${KAFKA_CLUSTER_ID}'  # Cluster ID generated externally
      KAFKA_SUPER_USERS: User:Admin;User:ANONYMOUS
    volumes:
      - ./kafka/kafka.jaas.conf:/etc/kafka/configs/kafka_server_jaas.conf

  # Kafka broker on the second node
  kafka2:
    image: confluentinc/cp-kafka:latest
    hostname: kafka2
    container_name: kafka2
    restart: always
    ports:
      - "59192:59092"  # External listener for client connections
      - "29094:29094"  # Controller listener
      - "19093:19093"  # Internal broker-to-broker listener
    environment:
      KAFKA_NODE_ID: 2
      KAFKA_ADVERTISED_LISTENERS: 'BROKER://${KAFKA_BROKER2_IP}:19093,EXTERNAL://${KAFKA_BROKER2_IP}:59192'
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@${KAFKA_BROKER1_IP}:29093,2@${KAFKA_BROKER2_IP}:29094'
      CLUSTER_ID: '${KAFKA_CLUSTER_ID}'
      KAFKA_SUPER_USERS: User:Admin;User:ANONYMOUS
    volumes:
      - ./kafka/kafka.jaas.conf:/etc/kafka/configs/kafka_server_jaas.conf

  # Cassandra instance on the first node
  cassandra1:
    image: cassandra:4.1.0
    container_name: cassandra1
    restart: always
    ports:
      - '9042:9042'  # Default Cassandra CQL port
      - '7000:7000'  # Inter-node communication
    environment:
      CASSANDRA_CLUSTER_NAME: MyCassandraCluster
      CASSANDRA_SEEDS: ${CASSANDRA_SEEDS}
      CASSANDRA_BROADCAST_ADDRESS: ${CASSANDRA_BROADCAST_ADDRESS1}
      CASSANDRA_LISTEN_ADDRESS: cassandra1
    volumes:
      - ./cassandra1_data:/var/lib/cassandra

  # Cassandra instance on the second node
  cassandra2:
    image: cassandra:4.1.0
    container_name: cassandra2
    restart: always
    ports:
      - '9142:9042'  # Default Cassandra CQL port for second node
      - '7100:7000'  # Inter-node communication for second node
    environment:
      CASSANDRA_CLUSTER_NAME: MyCassandraCluster
      CASSANDRA_SEEDS: ${CASSANDRA_SEEDS}
      CASSANDRA_BROADCAST_ADDRESS: ${CASSANDRA_BROADCAST_ADDRESS2}
      CASSANDRA_LISTEN_ADDRESS: cassandra2
    volumes:
      - ./cassandra2_data:/var/lib/cassandra

  # Flink JobManager for the first node
  jobmanager1:
    image: custom-flink-image  # Custom Flink image to be built
    container_name: jobmanager1
    restart: always
    ports:
      - "8081:8081"  # REST API
      - "46987:46987"  # RPC port
      - "6124:6124"  # Blob server port
      - "41435:41435"  # Metrics query service port
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        high-availability.type: zookeeper
        high-availability.storageDir: /opt/flink/recovery
        high-availability.zookeeper.quorum: ${ZOOKEEPER1_IP}:2181,${ZOOKEEPER2_IP}:2181,${ZOOKEEPER3_IP}:2182
        high-availability.zookeeper.path.root: /flink
        high-availability.cluster-id: /cluster
        jobmanager.rpc.address: ${JOBMANAGER1_IP}
        jobmanager.rpc.port: 6123
        blob.server.port: 6124
        rest.address: ${JOBMANAGER1_IP}
    volumes:
      - recovery:/opt/flink/recovery

  # Flink TaskManager for the first node
  taskmanager1:
    image: custom-flink-image
    container_name: taskmanager1
    restart: always
    depends_on:
      - jobmanager1
    command: taskmanager
    ports:
      - "50101:50101"  # Data port
      - "46989:46989"  # RPC port
    environment:
      - |
        FLINK_PROPERTIES=
        high-availability.type: zookeeper
        high-availability.storageDir: /opt/flink/recovery
        high-availability.zookeeper.quorum: ${ZOOKEEPER1_IP}:2181,${ZOOKEEPER2_IP}:2181,${ZOOKEEPER3_IP}:2182
        jobmanager.rpc.address: ${JOBMANAGER1_IP}
        taskmanager.host: ${TASKMANAGER1_IP}
        taskmanager.numberOfTaskSlots: 8
    volumes:
      - recovery:/opt/flink/recovery

  # Flink JobManager for the second node
  jobmanager2:
    image: custom-flink-image  # Custom Flink image to be built
    container_name: jobmanager2
    restart: always
    ports:
      - "8082:8081"  # REST API
      - "46988:46987"  # RPC port
      - "6125:6124"  # Blob server port
      - "41436:41435"  # Metrics query service port
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        high-availability.type: zookeeper
        high-availability.storageDir: /opt/flink/recovery
        high-availability.zookeeper.quorum: ${ZOOKEEPER1_IP}:2181,${ZOOKEEPER2_IP}:2181,${ZOOKEEPER3_IP}:2182
        high-availability.zookeeper.path.root: /flink
        high-availability.cluster-id: /cluster
        jobmanager.rpc.address: ${JOBMANAGER2_IP}
        jobmanager.rpc.port: 6123
        blob.server.port: 6124
        rest.address: ${JOBMANAGER2_IP}
    volumes:
      - recovery:/opt/flink/recovery

  # Flink TaskManager for the second node
  taskmanager2:
    image: custom-flink-image
    container_name: taskmanager2
    restart: always
    depends_on:
      - jobmanager2
    command: taskmanager
    ports:
      - "50102:50101"  # Data port
      - "46990:46989"  # RPC port
    environment:
      - |
        FLINK_PROPERTIES=
        high-availability.type: zookeeper
        high-availability.storageDir: /opt/flink/recovery
        high-availability.zookeeper.quorum: ${ZOOKEEPER1_IP}:2181,${ZOOKEEPER2_IP}:2181,${ZOOKEEPER3_IP}:2182
        jobmanager.rpc.address: ${JOBMANAGER2_IP}
        taskmanager.host: ${TASKMANAGER2_IP}
        taskmanager.numberOfTaskSlots: 8
    volumes:
      - recovery:/opt/flink/recovery
  
  zoo1:
    image: zookeeper
    restart: always
    hostname: zoo1
    ports:
      - 2181:2181
      - 2888:2888
      - 3888:3888
    environment:
      ZOO_MY_ID: 1
      ZOO_SERVERS: server.1=0.0.0.0:2888:3888;2181 server.2=${ZOOKEEPER2_IP}:2888:3888;2181 server.3=${ZOOKEEPER3_IP}:2889:3889;2182

  # Zookeeper Node 2
  zoo2:
    image: zookeeper
    restart: always
    hostname: zoo2
    ports:
      - 2181:2181
      - 2888:2888
      - 3888:3888
    environment:
      ZOO_MY_ID: 2
      ZOO_SERVERS: server.1=${ZOOKEEPER1_IP}:2888:3888;2181 server.2=0.0.0.0:2888:3888;2181 server.3=${ZOOKEEPER3_IP}:2889:3889;2182

  # Zookeeper Node 3
  zoo3:
    image: zookeeper
    restart: always
    hostname: zoo3
    ports:
      - 2182:2181
      - 2889:2888
      - 3889:3888
    environment:
      ZOO_MY_ID: 3
      ZOO_SERVERS: server.1=${ZOOKEEPER1_IP}:2888:3888;2181 server.2=${ZOOKEEPER2_IP}:2888:3888;2181 server.3=0.0.0.0:2889:3889;2182
      # ksqlDB Server 1
  ksqldb-server1:
    image: confluentinc/ksqldb-server:latest
    container_name: ksqldb-server1
    restart: always
    hostname: ksqldb-server1
    ports:
      - 8088:8088
    environment:
      KSQL_CONFIG_DIR: "/etc/ksqldb"
      KSQL_LISTENERS: "http://0.0.0.0:8088"
      KSQL_BOOTSTRAP_SERVERS: "${KAFKA_BROKER1_IP}:9092,${KAFKA_BROKER2_IP}:9092"
      KSQL_KSQL_SERVICE_ID: "ksqldb_service1"

  # ksqlDB Server 2
  ksqldb-server2:
    image: confluentinc/ksqldb-server:latest
    container_name: ksqldb-server2
    restart: always
    hostname: ksqldb-server2
    ports:
      - 8089:8088
    environment:
      KSQL_CONFIG_DIR: "/etc/ksqldb"
      KSQL_LISTENERS: "http://0.0.0.0:8088"
      KSQL_BOOTSTRAP_SERVERS: "${KAFKA_BROKER1_IP}:9092,${KAFKA_BROKER2_IP}:9092"
      KSQL_KSQL_SERVICE_ID: "ksqldb_service2"

volumes:
  recovery:
    driver: local
    driver_opts:
      type: none
      o: bind
      device: ./recovery