docker-compose-enterprise.yml

# Common kafka broker environment variables
x-kafka-broker-env: &kafka-broker-env
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT'
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 2
  KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
  KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
  KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 2
  KAFKA_JMX_HOSTNAME: localhost
  KAFKA_ENABLE_KRAFT: yes
  KAFKA_PROCESS_ROLES: 'broker,controller'
  KAFKA_CONTROLLER_QUORUM_VOTERS: '1@broker:29093,2@broker-2:29093'
  KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
  KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
  KAFKA_LOG_DIRS: '/tmp/kraft-combined-logs'
  KAFKA_METADATA_LOG_DIR: /tmp/kraft-metadata-logs
  # Replace CLUSTER_ID with a unique base64 UUID using "bin/kafka-storage.sh random-uuid"
  # See https://docs.confluent.io/kafka/operations-tools/kafka-tools.html#kafka-storage-sh
  CLUSTER_ID: 'QTkwRDkzMDg3QTdFNDJCRU'


# Common kafka broker configuration
x-kafka-broker-common: &kafka-broker-common
  image: confluentinc/cp-kafka:7.7.0
  extra_hosts:
    - "host.docker.internal:host-gateway"

# Common environment variables
x-kafka-connect-env: &kafka-connect-env
  CONNECT_BOOTSTRAP_SERVERS: 'broker:29092,broker-2:29092'
  CONNECT_GROUP_ID: compose-connect-group
  CONNECT_OFFSET_FLUSH_INTERVAL_MS: 500
  CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
  CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 2
  CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
  CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 2
  CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
  CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 2
  CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
  CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
  CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
  # CLASSPATH required due to CC-2422
  CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-7.5.3-0.jar
  CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
  CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
  CONNECT_ERRORS_RETRY_TIMEOUT: 90000
  CONNECT_ERRORS_RETRY_DELAY_MAX_MS: 120000
  CONNECT_PLUGIN_PATH: "/usr/share/java,/etc/kafka-connect/jars,/usr/share/confluent-hub-components"
  QUESTDB_HTTP_ENDPOINT: "${QUESTDB_HTTP_ENDPOINT:-host.docker.internal:9000}"
  QUESTDB_REST_TOKEN: "${QUESTDB_REST_TOKEN:-}"

# Common configuration (volumes, depends_on, etc.)
x-kafka-connect-common: &kafka-connect-common
  image: confluentinc/cp-kafka-connect:7.7.0
  depends_on:
    - broker-1
    - broker-2
  extra_hosts:
    - "host.docker.internal:host-gateway"
  volumes:
    - ./kafka-connect-plugins:/etc/kafka-connect/jars


services:
  grafana:
    image: grafana/grafana-oss:11.2.0
    container_name: rta_grafana
    restart: always
    user: "${DOCKER_COMPOSE_USER_ID:-}"
    ports:
      - 3000:3000
    extra_hosts:
      - "host.docker.internal:host-gateway"
    volumes:
      - ./dashboard/grafana/home_dir/var_lib_grafana:/var/lib/grafana/:rw
      - ./dashboard/grafana/home_dir/etc_grafana:/etc/grafana/:rw
    environment:
      - GF_INSTALL_PLUGINS=questdb-questdb-datasource
      - QDB_CLIENT_HOST=${QDB_CLIENT_HOST:-host.docker.internal}
      - QDB_CLIENT_PORT=${QDB_CLIENT_PORT:-8812}
      - QDB_CLIENT_USER=${QDB_CLIENT_USER:-admin}
      - QDB_CLIENT_PASSWORD=${QDB_CLIENT_PASSWORD:-quest}
      # use the value "disable" for local installations, and "require" for QuestDB Cloud
      - QDB_SSL_MODE=${QDB_SSL_MODE:-disable}

  jupyter-notebook:
    image:    jupyter/scipy-notebook
    container_name: rta_jupyter
    volumes:
      - ./notebooks:/home/jovyan:rw
      - ./ingestion/python:/home/ingestion:rw
      - ${QUESTDB_ROOT:-/var/tmp}:/db_root:ro
    ports:
      - 8888:8888
    command: sh -c "pip install questdb psycopg[binary] psycopg2-binary PyGithub kafka-python confluent-kafka \"confluent-kafka[avro]\" fastavro requests && start-notebook.sh --NotebookApp.password= --NotebookApp.token="
    extra_hosts:
      - "host.docker.internal:host-gateway"
    environment:
      - RESTARTABLE=yes
      - DOCKER_STACKS_JUPYTER_CMD=notebook
      - GITHUB_TOKEN=${GITHUB_TOKEN:-}
      - QUESTDB_HTTP_ENDPOINT=${QUESTDB_HTTP_ENDPOINT:-host.docker.internal:9000}
      - QUESTDB_REST_TOKEN=${QUESTDB_REST_TOKEN:-}
      - QDB_CLIENT_HOST=${QDB_CLIENT_HOST:-host.docker.internal}
      - QDB_CLIENT_PORT=${QDB_CLIENT_PORT:-8812}
      - QDB_CLIENT_USER=${QDB_CLIENT_USER:-admin}
      - QDB_CLIENT_PASSWORD=${QDB_CLIENT_PASSWORD:-quest}

  broker-1:
    <<: *kafka-broker-common
    hostname: broker
    container_name: rta_kafka_broker
    ports:
      - "9092:9092"
      - "9101:9101"
      - "8778:8778"   #Jolokia agent for MX monitoring metrics
    extra_hosts:
      - "host.docker.internal:host-gateway"
    volumes:
      - ./monitoring/kafka-agent:/usr/jolokia/agents:ro   #Jolokia agent for MX monitoring metrics
      - ./broker-1/kafka-data:/var/lib/kafka/data
      - ./broker-1/kafka-secrets:/etc/kafka/secrets
      - ./broker-1/tmp:/tmp
    environment:
      <<: *kafka-broker-env
      KAFKA_NODE_ID: 1
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092'
      KAFKA_JMX_PORT: 9101
      KAFKA_LISTENERS: 'PLAINTEXT://broker:29092,CONTROLLER://broker:29093,PLAINTEXT_HOST://0.0.0.0:9092'
      KAFKA_OPTS: -javaagent:/usr/jolokia/agents/jolokia-jvm-1.7.2.jar=port=8778,host=0.0.0.0    #Jolokia agent for MX monitoring metrics

  broker-2:
    <<: *kafka-broker-common
    hostname: broker-2
    container_name: rta_kafka_broker_2
    ports:
      - "9093:9093"
      - "9102:9102"
      - "8779:8779"   #Jolokia agent for MX monitoring metrics
    extra_hosts:
      - "host.docker.internal:host-gateway"
    volumes:
      - ./monitoring/kafka-agent:/usr/jolokia/agents:ro   #Jolokia agent for MX monitoring metrics
      - ./broker-2/kafka-data:/var/lib/kafka/data
      - ./broker-2/kafka-secrets:/etc/kafka/secrets
      - ./broker-2/tmp:/tmp
    environment:
      <<: *kafka-broker-env
      KAFKA_NODE_ID: 2
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://broker-2:29092,PLAINTEXT_HOST://localhost:9093'
      KAFKA_JMX_PORT: 9102
      KAFKA_LISTENERS: 'PLAINTEXT://broker-2:29092,CONTROLLER://broker-2:29093,PLAINTEXT_HOST://0.0.0.0:9093'
      KAFKA_OPTS: -javaagent:/usr/jolokia/agents/jolokia-jvm-1.7.2.jar=port=8779,host=0.0.0.0  #Jolokia agent for MX monitoring metrics

  schema_registry:
    image: confluentinc/cp-schema-registry:7.7.0
    hostname: schema_registry
    container_name: rta_schema_registry
    depends_on:
      - broker-1
      - broker-2
    ports:
      - "8081:8081"
    extra_hosts:
      - "host.docker.internal:host-gateway"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema_registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092,broker-2:29092'
      SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_ORIGIN: '*'
      SCHEMA_REGISTRY_ACCESS_CONTROL_ALLOW_METHODS: 'GET,POST,PUT,OPTIONS'

  kafka-connect-1:
    <<: *kafka-connect-common
    hostname: kafka-connect
    container_name: rta_kafka_connect
    ports:
      - "8083:8083"
    environment:
      <<: *kafka-connect-env
      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect
      CONNECT_LISTENERS: http://0.0.0.0:8083
    command:
    - bash
    - -c
    - |
      # Launch Kafka Connect
      /etc/confluent/docker/run &
      #
      # Wait for Kafka Connect listener
      echo "Waiting for Kafka Connect to start listening on localhost ⏳"
      while : ; do
        curl_status=$$(curl -s -o /dev/null -w %{http_code} http://localhost:8083/connectors)
        echo -e $$(date) " Kafka Connect listener HTTP state: " $$curl_status " (waiting for 200)"
        if [ $$curl_status -eq 200 ] ; then
          break
        fi
        sleep 5
      done

      echo -e "\n--\n+> Registering QuestDB Connector"

      curl -s -X PUT -H  "Content-Type:application/json" http://localhost:8083/connectors/questdb-iot/config -d '{
          "connector.class": "io.questdb.kafka.QuestDBSinkConnector",
          "tasks.max": "5",
          "topics": "iot_data",
          "client.conf.string": "https::addr=${QUESTDB_HTTP_ENDPOINT:-host.docker.internal:9000};token=${QUESTDB_REST_TOKEN:-};tls_verify=unsafe_off;auto_flush_interval=500;",
          "name": "questdb-iot",
          "value.converter": "org.apache.kafka.connect.json.JsonConverter",
          "include.key": false,
          "key.converter": "org.apache.kafka.connect.storage.StringConverter",
          "table": "iot_data",
          "symbols": "device_type",
          "value.converter.schemas.enable": false
      }'

      curl -s -X PUT -H  "Content-Type:application/json" http://localhost:8083/connectors/questdb-github/config -d '{
          "connector.class": "io.questdb.kafka.QuestDBSinkConnector",
          "tasks.max": "5",
          "topics": "github_events",
          "client.conf.string": "https::addr=${QUESTDB_HTTP_ENDPOINT:-host.docker.internal:9000};token=${QUESTDB_REST_TOKEN:-};tls_verify=unsafe_off;auto_flush_interval=500;",
          "name": "questdb-github",
          "value.converter": "org.apache.kafka.connect.json.JsonConverter",
          "include.key": false,
          "key.converter": "org.apache.kafka.connect.storage.StringConverter",
          "table": "github_events",
          "symbols": "type,repo",
          "timestamp.field.name": "created_at",
          "value.converter.schemas.enable": false
      }'

      curl -s -X PUT -H  "Content-Type:application/json" http://localhost:8083/connectors/questdb-trades/config -d '{
          "connector.class": "io.questdb.kafka.QuestDBSinkConnector",
          "tasks.max": "5",
          "topics": "trades",
          "client.conf.string": "https::addr=${QUESTDB_HTTP_ENDPOINT:-host.docker.internal:9000};token=${QUESTDB_REST_TOKEN:-};tls_verify=unsafe_off;auto_flush_interval=500;",
          "name": "questdb-trades",
          "value.converter": "io.confluent.connect.avro.AvroConverter",
          "value.converter.schema.registry.url": "http://schema_registry:8081",
          "include.key": false,
          "key.converter": "io.confluent.connect.avro.AvroConverter",
          "key.converter.schema.registry.url": "http://schema_registry:8081",
          "table": "trades",
          "symbols": "symbol, side",
          "timestamp.field.name": "timestamp",
          "value.converter.schemas.enable": true
      }'

      curl -s -X PUT -H  "Content-Type:application/json" http://localhost:8083/connectors/questdb-smart-meters/config -d '{
          "connector.class": "io.questdb.kafka.QuestDBSinkConnector",
          "tasks.max": "5",
          "topics": "smart-meters",
          "client.conf.string": "https::addr=${QUESTDB_HTTP_ENDPOINT:-host.docker.internal:9000};token=${QUESTDB_REST_TOKEN:-};tls_verify=unsafe_off;auto_flush_interval=500;",
          "name": "questdb-smart-meters",
          "value.converter": "io.confluent.connect.avro.AvroConverter",
          "value.converter.schema.registry.url": "http://schema_registry:8081",
          "include.key": false,
          "key.converter": "io.confluent.connect.avro.AvroConverter",
          "key.converter.schema.registry.url": "http://schema_registry:8081",
          "table": "smart_meters",
          "symbols": "device_id, mark_model, status",
          "timestamp.field.name": "timestamp",
          "value.converter.schemas.enable": true
      }'

      curl -s -X PUT -H  "Content-Type:application/json" http://localhost:8083/connectors/questdb-transactions/config -d '{
          "connector.class": "io.questdb.kafka.QuestDBSinkConnector",
          "tasks.max": "5",
          "topics": "transactions",
          "client.conf.string": "https::addr=${QUESTDB_HTTP_ENDPOINT:-host.docker.internal:9000};token=${QUESTDB_REST_TOKEN:-};tls_verify=unsafe_off;auto_flush_interval=500;",
          "name": "questdb-transactions",
          "value.converter": "io.confluent.connect.avro.AvroConverter",
          "value.converter.schema.registry.url": "http://schema_registry:8081",
          "include.key": false,
          "key.converter": "io.confluent.connect.avro.AvroConverter",
          "key.converter.schema.registry.url": "http://schema_registry:8081",
          "table": "transactions",
          "symbols": "merchant, category, gender, city, state",
          "timestamp.field.name": "timestamp",
          "value.converter.schemas.enable": true
      }'

      sleep infinity

  kafka-connect-2:
    <<: *kafka-connect-common
    hostname: kafka-connect-2
    container_name: rta_kafka_connect_2
    ports:
      - "8084:8084"
    environment:
      <<: *kafka-connect-env
      CONNECT_REST_ADVERTISED_HOST_NAME: kafka-connect-2
      CONNECT_LISTENERS: http://0.0.0.0:8084



        #  telegraf:
        #    image: telegraf
        #    container_name: rta_telegraf
        #    depends_on:
        #      - broker-1
        #      - broker-2
        #    volumes:
        #      - ./monitoring/telegraf/kafka_and_questdb_telegraf.conf:/etc/telegraf/telegraf.conf:ro
        #    extra_hosts:
        #      - "host.docker.internal:host-gateway"
        #    environment:
        #      - QUESTDB_METRICS_URL=${QUESTDB_METRICS_URL:-https://172.31.42.41:9003}
        #      - QUESTDB_HTTP_URL=${QUESTDB_HTTP_URL:-https://172.31.42.41:9000}
        #      - QUESTDB_HTTP_TOKEN=${QUESTDB_HTTP_TOKEN:-qt1H8Mi_vPUpt_R7ByDdJ7_kRIiieNb_BFKdjBDqxp6wh8}
        #      - QUESTDB_SKIP_TLS_VERIFICATION=${QUESTDB_SKIP_TLS_VERIFICATION:-true}