Docker Compose环境下无法向Kafka Topic发送消息问题排查

云原生阿岚

2026-8-13

问题描述

我用docker-compose.yml启动并关联多个服务，现有一个Jar包需向容器内的Kafka Topic发送JSON格式消息，但偶尔发送失败，报错如下：

WARNING: Error sending message with key 1,667,770,078,000
Topic general-events not present in metadata after 60000 ms.

docker-compose配置

version: "3"

services:
  zookeeper:
    image: "bitnami/zookeeper:latest"
    container_name: zookeeper
    hostname: zookeeper
    ports:
      - "2181:2181"
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes

  kafka:
    image: confluentinc/cp-kafka:6.1.1
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - '9092:9092'
    expose:
      - '29092'
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: '1'
      KAFKA_MIN_INSYNC_REPLICAS: '1'


  init-kafka:
    image: confluentinc/cp-kafka:6.1.1
    container_name: init-kafka
    depends_on:
      - kafka
      - zookeeper
    entrypoint: [ '/bin/sh', '-c' ]
    command: |
      "
      # blocks until kafka is reachable
      kafka-topics --bootstrap-server kafka:29092 --list

      echo -e 'Creating kafka topics'
      kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic general-events --replication-factor 1 --partitions 1

      echo -e 'Successfully created the following topics:'
      kafka-topics --bootstrap-server kafka:29092 --list

      kafka-console-consumer --topic general-events --from-beginning --bootstrap-server kafka:29092
      "
    
  influxdb:
    image: influxdb
    container_name: influxdb
    hostname: influxdb
    volumes:
      - influxdb-storage:/var/lib/influxdb2:rw
    env_file:
      - .env
    entrypoint: ["./entrypoint.sh"]
    ports:
      - ${DOCKER_INFLUXDB_INIT_PORT}:8086

  telegraf:
    image: telegraf
    depends_on:
      - influxdb
      - kafka
    container_name: telegraf
    links:
      - influxdb
    restart: on-failure
    env_file:
      - .env
    environment: 
      - DOCKER_INFLUXDB_INIT_ORG=earthWatch
      - DOCKER_INFLUXDB_INIT_BUCKET=telegraf
      - DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=60b42f6f12a91425b4fc02d1dd4e44eff9231f737171da79a993055c3aa367ab
    volumes:
      - ./telegraf/telegraf.conf:/etc/telegraf/telegraf.conf:rw

  java:
    image: openjdk:15
    depends_on:
      - init-kafka
    container_name: data-source
    volumes:
      - ./earthWatch.jar:/usr/src/java 
    command: bash -c "java -jar /usr/src/java earthWatch.jar"

volumes:
  influxdb-storage:

Producer类配置

public class Producer implements Runnable {

    private static final Logger LOGGER = Logger.getLogger(Producer.class.getName());
    private static final String TOPIC_NAME = "general-events";
    private KafkaProducer<Long, String> kafkaProducer = null;
    private final String KAFKA_CLUSTER_ENV_VAR_NAME = "KAFKA_CLUSTER";

    public Producer() {
        LOGGER.log(Level.INFO, "Kafka Producer running in thread {0}", Thread.currentThread().getName());
        Properties kafkaProps = new Properties();

        String defaultClusterValue = "172.21.0.4:29092";
        String kafkaCluster = System.getProperty(KAFKA_CLUSTER_ENV_VAR_NAME, defaultClusterValue);
        LOGGER.log(Level.INFO, "Kafka cluster {0}", kafkaCluster);

        kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCluster);
        kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.LongSerializer");
        kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProps.put(ProducerConfig.ACKS_CONFIG, "0");

        this.kafkaProducer = new KafkaProducer<>(kafkaProps);

    }

日志信息

[2022-11-06 21:35:13,759] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:13,866] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:13,969] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:14,176] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:14,578] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:15,383] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:16,591] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:17,795] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-11-06 21:35:18,900] WARN [AdminClient clientId=adminclient-1] Connection to node -1 (kafka/172.21.0.5:29092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

Creating kafka topics
Created topic general-events.
Successfully created the following topics:
general-events

已确认Topic创建成功，但最终没有消息被消费，请问可能是什么问题？

问题排查与解决建议

1. Kafka地址硬编码问题

Producer中使用固定IP 172.21.0.4:29092 作为默认地址，Docker容器重启后IP可能变化，导致连接失败。

解决：在docker-compose的Java服务中添加环境变量，用Kafka容器名作为地址（Docker内部DNS可解析容器名）：

java:
  # 其他配置不变
  environment:
    - KAFKA_CLUSTER=kafka:29092

同时修改Producer代码，优先读取系统环境变量（原代码用System.getProperty读取的是JVM属性，不是容器环境变量）：

String kafkaCluster = System.getenv(KAFKA_CLUSTER_ENV_VAR_NAME);
if (kafkaCluster == null || kafkaCluster.isEmpty()) {
    kafkaCluster = "kafka:29092"; // 用容器名作为默认值，而非固定IP
}

2. Kafka就绪判断不严谨

depends_on仅保证容器启动顺序，不保证Kafka服务完全就绪。init-kafka虽在Kafka之后启动，但Kafka可能未完成初始化，导致前期连接失败；且init-kafka中启动的消费者会让容器一直运行，Java服务会持续等待无法启动。

解决：修改init-kafka的命令，添加循环等待Kafka就绪，并移除消费者命令：

"
# 循环等待Kafka可用
until kafka-topics --bootstrap-server kafka:29092 --list > /dev/null 2>&1; do
    echo 'Waiting for Kafka to be ready...'
    sleep 2
done

echo -e 'Creating kafka topics'
kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic general-events --replication-factor 1 --partitions 1

echo -e 'Successfully created the following topics:'
kafka-topics --bootstrap-server kafka:29092 --list
"

3. Producer的ACK配置风险

当前设置ACKS_CONFIG: "0"，意味着Producer发送消息后不等待任何确认，消息可能直接丢失且无报错；若Kafka未完全就绪，消息丢失概率会更高。

建议：测试阶段改为acks=1，确保消息被Broker接收后再返回，便于排查问题；生产环境根据业务需求调整，但不建议用0除非能接受消息丢失。

4. 容器网络连通性验证

检查Java容器与Kafka容器是否在同一Docker网络（默认docker-compose会创建默认网络，所有服务都在其中，但需确认无自定义网络冲突）。

验证：进入Java容器执行ping kafka，若无法解析IP，需检查网络配置。

5. 消费者配置排查

若依赖init-kafka中的消费者，该消费者仅在init-kafka启动时运行，后续可能已退出，无法接收新消息；
检查telegraf的telegraf.conf，确认是否正确配置了general-events Topic的消费者；
临时验证：在主机执行kafka-console-consumer --bootstrap-server localhost:9092 --topic general-events --from-beginning，或进入Kafka容器执行kafka-console-consumer --bootstrap-server kafka:29092 --topic general-events --from-beginning，确认是否能接收到消息，以此区分是Producer还是消费者的问题。

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

Docker Compose环境下无法向Kafka Topic发送消息问题排查

docker-compose配置 #

Producer类配置 #

日志信息 #

1. Kafka地址硬编码问题 #

2. Kafka就绪判断不严谨 #

3. Producer的ACK配置风险 #

4. 容器网络连通性验证 #

5. 消费者配置排查 #