flink 1.20 and kafka and SR in docker

docker-compose.yaml

services:
  kafka:
    image: apache/kafka:3.8.0
    hostname: kafka
    ports:
      - "9092:9092"
      - "29092:29092"
    healthcheck:
      test: ["CMD-SHELL", "kafka-topics.sh --bootstrap-server kafka:9092 --list"]
      interval: 5s
      timeout: 10s
      retries: 10
    environment:
      - PATH=/opt/kafka/bin:$PATH
      # KRaft settings
      - KAFKA_NODE_ID=0
      - KAFKA_PROCESS_ROLES=controller,broker
      - KAFKA_CONTROLLER_QUORUM_VOTERS=0@kafka:9093
      # Listeners
      - KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093,PLAINTEXT_HOST://:29092
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      - KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT
      # Additional settings
      - KAFKA_AUTO_CREATE_TOPICS_ENABLE=true
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1
      - KAFKA_MIN_INSYNC_REPLICAS=1

  schema-registry:
    image: confluentinc/cp-schema-registry:7.9.0
    hostname: schema-registry
    ports:
      - "8081:8081"
    depends_on:
      kafka:
        condition: service_healthy
    environment:
      - SCHEMA_REGISTRY_HOST_NAME=schema-registry
      - SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=kafka:9092
      - SCHEMA_REGISTRY_LISTENERS=http://0.0.0.0:8081
    healthcheck:
      test: ["CMD-SHELL", "curl -f http://localhost:8081 || exit 1"]
      interval: 5s
      timeout: 10s
      retries: 10

  # Flink components
  jobmanager:
    image: flink:1.20.0
    hostname: jobmanager
    ports:
      - "8082:8081"  # Already set correctly - external 8082, internal 8081
    command: jobmanager
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8081/overview"]
      interval: 5s
      timeout: 10s
      retries: 10

  taskmanager:
    image: flink:1.20.0
    depends_on:
      jobmanager:
        condition: service_healthy
    command: taskmanager
    scale: 1
    environment:
      - |
        FLINK_PROPERTIES=
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 16

  sql-client:
    build:
      context: ./flink-sql/docker/sql-client
    command:
      - bin/sql-client.sh
    depends_on:
      jobmanager:
        condition: service_healthy
      kafka:
        condition: service_healthy
      schema-registry:
        condition: service_healthy
    volumes:
      - ./flink-sql/sql:/opt/sql
    environment:
      - FLINK_JOBMANAGER_HOST=jobmanager
      - FLINK_JOBMANAGER_PORT=8081
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092
      - SCHEMA_REGISTRY_URL=http://schema-registry:8081
      - |
        - FLINK_PROPERTIES=
          jobmanager.rpc.address: jobmanager
          rest.address: jobmanager

Dockerfile

FROM flink:1.20.0

# Install necessary utilities
RUN apt-get update && \
    apt-get install -y --no-install-recommends \
    curl \
    wget \
    jq \
    netcat-openbsd \
    && rm -rf /var/lib/apt/lists/*

# Add Flink connector dependencies
RUN mkdir -p /opt/flink/lib

# Add necessary connectors for SQL Client
# Kafka connector
RUN wget -P /opt/flink/lib https://repo.maven.apache.org/maven2/org/apache/flink/flink-connector-kafka/3.4.0-1.20/flink-connector-kafka-3.4.0-1.20.jar
RUN wget -P /opt/flink/lib https://repo.maven.apache.org/maven2/org/apache/kafka/kafka-clients/3.8.0/kafka-clients-3.8.0.jar

# Avro and Schema Registry support
RUN wget -P /opt/flink/lib https://repo.maven.apache.org/maven2/org/apache/flink/flink-avro/1.20.0/flink-avro-1.20.0.jar
RUN wget -P /opt/flink/lib https://packages.confluent.io/maven/io/confluent/kafka-avro-serializer/7.9.0/kafka-avro-serializer-7.9.0.jar
RUN wget -P /opt/flink/lib https://packages.confluent.io/maven/io/confluent/kafka-schema-registry-client/7.9.0/kafka-schema-registry-client-7.9.0.jar
RUN wget -P /opt/flink/lib https://packages.confluent.io/maven/io/confluent/common-utils/7.9.0/common-utils-7.9.0.jar

# JSON format
RUN wget -P /opt/flink/lib https://repo.maven.apache.org/maven2/org/apache/flink/flink-json/1.20.0/flink-json-1.20.0.jar

RUN chown -R flink:flink /opt/flink/lib

test SELECT CURRENT_TIMESTAMP, RAND() FROM (VALUES (1)) as t(n);