Kafka生产者消息发送超时、消费者无法接收问题求助

问题现象 #

• 生产者发送消息超时，报错信息：

Expiring 2 record(s) for catering-0:120012 ms has passed since batch creation

• 消费者完全接收不到消息
• 应用启动后第一分钟会出现以下连接错误，之后显示连接成功：

[AdminClient clientId=adminclient-1] Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.

配置变更过程 #

初始配置（wurstmeister镜像） #

因无法同时通过localhost和容器内部连接Kafka，更换了镜像及配置。初始docker-compose配置如下：

zookeeper:
  image: wurstmeister/zookeeper
  container_name: zookeeper
  ports:
    - "2181:2181"
  restart: always
  
kafka:
  image: wurstmeister/kafka
  container_name: kafka
  ports:
    - "9092:9092"
  depends_on:
    - zookeeper
  restart: always
  environment:
    # KAFKA_ADVERTISED_HOST_NAME: kafka
    KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
    KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    KAFKA_CREATE_TOPICS: "test_topic:1:3"

修改后配置（confluentinc镜像） #

更换为confluentinc镜像后的docker-compose配置：

zookeeper1:
  image: confluentinc/cp-zookeeper
  container_name: zookeeper
  ports:
    - "2181:2181"
  restart: always
  environment:
    ZOOKEEPER_CLIENT_PORT: 2181
    ZOOKEEPER_TICK_TIME: 2000
    ZOOKEEPER_SYNC_LIMIT: 2

kafka1:
  image: confluentinc/cp-kafka
  container_name: kafka
  ports:
    - target: 9094
      published: 9094
      protocol: tcp
      mode: host
  depends_on:
    - zookeeper1
  environment:
    KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
    KAFKA_LISTENERS: INTERNAL://0.0.0.0:9092,OUTSIDE://0.0.0.0:9094
    KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,OUTSIDE://localhost:9094
    KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,OUTSIDE:PLAINTEXT
    KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
    KAFKA_CREATE_TOPICS: "catering:1:3"

仅修改了消费者和生产者的端口配置，但问题依旧；回退至初始配置后，问题仍未解决。

生产者配置 #

kafka:
  producer:
    bootstrap-servers: localhost:9094
    key-serializer: org.apache.kafka.common.serialization.StringSerializer
    value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

消费者配置 #

kafka:
  consumer:
    bootstrap-servers: localhost:9094
    group-id: json
    auto-offset-reset: earliest
    key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer

恳请帮忙排查解决该问题。

内容的提问来源于stack exchange，提问作者Орорп Орол