Docker化Kafka跨容器消费无消息问题排查及解决方法

问题背景 #

我正尝试让同一主机上不同Docker容器中的Kafka生产者与消费者实现通信，目前两者都运行在本地开发机器上，最终目标是让Docker化的消费者运行在远程服务器上。

现在的问题是：当消费者运行在独立容器中时，无法接收到任何消息。我期望容器化的消费者能够接收并打印来自生产者的消息。

已验证情况 #

• 与生产者同容器的消费者可以接收消息
• 主机上非容器化的消费者也能接收消息
• 尝试将两个容器置于同一Docker网络中，但仍无消息接收

Docker化消费者能正常创建并运行，但没有任何消息输出，处于静默状态。

相关配置 #

生产者及关联服务Docker Compose配置 #

kafka:
  image: quay.io/debezium/kafka:2.3
  ports:
    - 9092:9092
  environment:
    NODE_ID: 1
    ZOOKEEPER_CONNECT: zookeeper:2181
    KAFKA_LISTENERS: INTERNAL://:29092,EXTERNAL://:9092
    KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
    KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
zookeeper:
  image: quay.io/debezium/zookeeper:2.3
  ports:
    - 2181:2181
    - 2888:2888
    - 3888:3888
apicurio:
  image: apicurio/apicurio-registry-mem:2.4.1.Final
  ports:
    - 8080:8080
connect:
  image: quay.io/debezium/connect:2.3
  ports:
    - 8083:8083
  environment:
    BOOTSTRAP_SERVERS: kafka:29092
    GROUP_ID: 1
    CONFIG_STORAGE_TOPIC: my_connect_configs
    OFFSET_STORAGE_TOPIC: my_connect_offsets
    STATUS_STORAGE_TOPIC: my_connect_statuses
    ENABLE_APICURIO_CONVERTERS: true
  links:
    - cdm-db
    - kafka
    - apicurio
  depends_on:
    - cdm-db
    - kafka
cdm-db:
  image: debezium/postgres:13-alpine
  ports:
    - 5433:5432
  volumes:
    - dev-db-data:/var/lib/postgresql/data
  environment:
    POSTGRES_DB: ${DB_NAME}
    POSTGRES_USER: user
    POSTGRES_PASSWORD: password

消费者应用Docker Compose配置 #

app:
  build:
    context: .
    dockerfile: docker/Dockerfile
  entrypoint: ["sh", "-c", "python -u manage.py runserver 0.0.0.0:8000"]
  volumes:
    - ./static:/srv/static
    - ./media:/srv/media
    - ./app:/srv/app
    - ./import/:/srv/import
  ports:
    - "127.0.0.1:${APP_PORT}:8000"
  extra_hosts:
    - "host.docker.internal:host-gateway"
  depends_on:
    - db
  links:
    - db
    - mq
    - redis
  environment:
    PYTHONUNBUFFERED: 1
  restart: always

消费者进程代码 #

from kafka.consumer import KafkaConsumer

#BOOTSTRAP_SERVERS="kafka:29092"    <----该配置在消费者与生产者同容器时生效
#BOOTSTRAP_SERVERS="localhost:9092" <----该配置在主机上非容器化消费者时生效
BOOTSTRAP_SERVERS="host.docker.internal:9092" <----该配置用于主机上独立容器的消费者，但创建后无输出

def main():
    print("Starting consumer", BOOTSTRAP_SERVERS)

    try:
        consumer = KafkaConsumer(
            bootstrap_servers=[BOOTSTRAP_SERVERS],
            )
    except Exception as error:
       print("Error creating KafkaConsumer", type(error).__name__, "-", error)
       return -1

    print("consumer created.")
    print(consumer.config)

    consumer.subscribe("dbserver1.public.geography_country")
    try:
        for msg in consumer:
            print("consuming")
            print(msg)
    except Exception as error:
        print("Error consuming", type(error).__name__, "-", error)

    print("done")

if __name__ == "__main__":
    main()

问题解决方法 #

通过修改Kafka服务的Docker Compose配置解决了问题：

原配置：

KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://localhost:9092 

修改为：

KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:29092,EXTERNAL://host.docker.internal:9092

内容的提问来源于stack exchange，提问作者mattjs