docker-compose.yml

version: "3"
services:

  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181

  kafka1:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka1
    ports:
      - "8097:8097"
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: EXTERNAL://localhost:8097,INTERNAL://kafka1:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2

  kafka2:
    image: confluentinc/cp-kafka:7.3.0
    container_name: kafka2
    ports:
      - "8098:8098"
    depends_on:
      - zookeeper
    environment:
      KAFKA_ADVERTISED_HOST_NAME: Kafka2
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: EXTERNAL://localhost:8098,INTERNAL://kafka2:9092
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2

  # Create topic directly in docker compose
  init-kafka:
    image: confluentinc/cp-kafka:7.3.0
    depends_on:
      - kafka1
      - kafka2
    entrypoint: [ '/bin/sh', '-c' ]
    command: |
      "  
      echo -e 'Creating kafka topics'
      kafka-topics --bootstrap-server kafka1:9092 --create --if-not-exists --topic raw_trafficdata --replication-factor 2 --partitions 2
      kafka-topics --bootstrap-server kafka1:9092 --create --if-not-exists --topic raw_weatherdata --replication-factor 2 --partitions 2
      kafka-topics --bootstrap-server kafka1:9092 --create --if-not-exists --topic processed_trafficdata --replication-factor 2 --partitions 2
      kafka-topics --bootstrap-server kafka1:9092 --create --if-not-exists --topic processed_weatherdata --replication-factor 2 --partitions 2
      
      echo -e 'Successfully created the following topics:'
      kafka-topics --bootstrap-server kafka1:9092 --list
      "

  schema-registry:
    image: confluentinc/cp-schema-registry:7.3.0
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "kafka1:9092,kafka2:9092"
      SCHEMA_REGISTRY_LISTENERS: "http://0.0.0.0:8081"
      SCHEMA_REGISTRY_HOST_NAME: "schema-registry"
      SCHEMA_REGISTRY_KAFKASTORE_TOPIC_REPLICATION_FACTOR: "2"
    depends_on:
      - "kafka1"
      - "kafka2"

  rest-proxy:
    image: confluentinc/cp-kafka-rest:7.3.0
    ports:
      - "8082:8082"
    environment:
      KAFKA_REST_BOOTSTRAP_SERVERS: "kafka1:9092,kafka2:9092"
      KAFKA_REST_LISTENERS: "http://0.0.0.0:8082/"
      KAFKA_REST_HOST_NAME: "rest-proxy"
      KAFKA_REST_SCHEMA_REGISTRY_URL: "http://schema-registry:8081/"
    depends_on:
      - "kafka1"
      - "kafka2"
      - "schema-registry"

  connect:
    image: confluentinc/cp-kafka-connect-base:7.3.0
    depends_on:
      - kafka1
      - kafka2
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: "kafka1:9092,kafka2:9092"
      CONNECT_REST_PORT: 8083
      CONNECT_GROUP_ID: compose-connect-group
      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_KEY_CONVERTER: "io.confluent.connect.avro.AvroConverter"
      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_VALUE_CONVERTER: "io.confluent.connect.avro.AvroConverter"
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONNECT_CONFIG_STORAGE_TOPIC: "connect-config"
      CONNECT_OFFSET_STORAGE_TOPIC: "connect-offset"
      CONNECT_STATUS_STORAGE_TOPIC: "connect-status"
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_LOG4J_ROOT_LOGLEVEL: "ERROR"
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: "2"
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: "2"
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: "2"
    command:
      - bash
      - -c
      - |
        echo "Installing Connector"
        confluent-hub install --no-prompt confluentinc/kafka-connect-jdbc:10.6.0
        
        #
        echo "Launching Kafka Connect worker"
        /etc/confluent/docker/run &
        #
        sleep infinity

  flask:
    image: flask_app
    build: flask_app
    ports:
      - "3030:3030"
    links:
      - kafka1
      - kafka2
      - producer
    environment:
      PYTHONUNBUFFERED: 1

  producer:
    image: producer_python
    build: ./producer
    ports:
      - "3031:3031"
    links:
      - kafka1
      - kafka2
    environment:
      PYTHONUNBUFFERED: 1
      DOCKER: True
    depends_on:
      - schema-registry
    entrypoint: [ '/bin/sh', '-c' ]
    command: |
      "
      /app/wait_for_services.sh
      cd produce_data
      python3 produce_rawdata.py
      "

  dashboard:
    image: dashboard
    build: ./dashboard
    command: "streamlit run streamlit.py"
    ports:
      - "8501:8501"
    volumes:
      - "./src:/usr/src/app/src"
    environment:
      PYTHONUNBUFFERED: 1
      DOCKER: True
    depends_on:
      - producer
      - kafka1

  postgres:
    image: postgres:15.1
    environment:
      - POSTGRES_DB=mydatabase
      - POSTGRES_USER=myuser
      - POSTGRES_PASSWORD=mypassword
    ports:
      - "5432:5432"
    volumes:
      - ./init.sql:/docker-entrypoint-initdb.d/init.sql
      - ./data:/var/lib/postgresql/data

  faust:
    image: producer_python
    build: ./producer
    ports:
      - "3032:3032"
    restart: on-failure
    depends_on:
      - zookeeper
      - kafka1
      - kafka2
      - schema-registry
    entrypoint: [ '/bin/sh', '-c' ]
    command: |
      "
      /app/wait_for_services.sh
      cd stream_data
      faust -A stream worker -l info
      "

  init_connector:
    image: producer_python
    build: ./producer
    ports:
      - "3033:3033"
    depends_on:
      - connect
    entrypoint: [ '/bin/sh', '-c' ]
    command: |
      "
      /app/wait_for_services.sh
      cd persist_data
      python3 connector.py
      "