本文旨在解决在使用 Docker Compose 部署 Kafka 集群时，应用程序无法向 Kafka 主题发送消息的问题。我们将分析常见的配置错误，并提供修改建议，确保生产者能够正确连接到 Kafka Broker，从而成功发送消息。通过调整 Kafka 的监听器配置以及生产者端的 Broker 地址，可以有效解决此类连接问题。
问题分析
在使用 Docker Compose 部署 Kafka 集群时，生产者无法发送消息到 Kafka 主题，并出现类似 Topic general-events not present in metadata after 60000 ms 的错误，通常是由于以下原因导致：
- Kafka Broker 地址配置错误： 生产者配置的 Broker 地址与 Kafka 实际监听的地址不匹配。
- 网络配置问题： 容器之间的网络连接存在问题，导致生产者无法访问 Kafka Broker。
- Kafka 监听器配置不正确： Kafka 的监听器配置可能不允许从容器外部或特定的网络访问。
解决方案
针对上述问题，可以采取以下步骤进行排查和解决：
1. 检查 Kafka 监听器配置
Kafka 的 KAFKA_ADVERTISED_LISTENERS 环境变量至关重要，它决定了 Kafka Broker 如何向客户端公布自己的地址。确保该配置正确反映了客户端访问 Kafka 的方式。
在 docker-compose.yml 文件中，检查 Kafka 服务的 environment 部分：
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'- PLAINTEXT://kafka:29092：允许容器内部的其他服务通过 kafka 这个 hostname 和 29092 端口访问 Kafka Broker。
- PLAINTEXT_HOST://localhost:9092：允许宿主机通过 localhost 和 9092 端口访问 Kafka Broker。
注意事项：
- 如果你的生产者运行在 Docker 容器之外，确保 PLAINTEXT_HOST 配置正确，并且宿主机能够访问 Kafka Broker。
- 如果你的生产者运行在 Docker 容器内部，但与 Kafka Broker 不在同一个 Docker 网络中，你需要配置一个可以从生产者容器访问的地址。
2. 修改生产者 Broker 地址
生产者需要使用正确的 Broker 地址才能成功连接到 Kafka 集群。如果 Kafka Broker 的地址发生变化，或者生产者使用了错误的地址，就会导致连接失败。
在生产者代码中，检查 bootstrap.servers 配置项。确保它指向正确的 Kafka Broker 地址。
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import java.util.Properties;
import java.util.logging.Level;
import java.util.logging.Logger;
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();
// 使用环境变量或默认值配置 Kafka Broker 地址
String defaultClusterValue = "kafka:29092"; // 修改为 kafka: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);
}
}注意事项：
- 如果生产者运行在 Docker 容器内部，应该使用 Kafka 容器的 hostname (kafka) 和内部端口 (29092)。
- 如果生产者运行在 Docker 容器外部，应该使用 localhost 和映射到宿主机的端口 (9092)。
3. 检查容器网络
确保生产者容器和 Kafka 容器在同一个 Docker 网络中。可以使用 docker network inspect <network_name> 命令来检查容器的网络配置。
如果生产者容器和 Kafka 容器不在同一个网络中，可以使用 docker network connect <network_name> <container_name> 命令将生产者容器连接到 Kafka 容器所在的网络。
4. 验证 Topic 创建
虽然 init-kafka 服务创建了 topic，但仍建议再次确认 topic 是否成功创建。可以在 Kafka 容器内部执行以下命令：
docker exec -it kafka bash kafka-topics --bootstrap-server kafka:29092 --list
确认列表中包含 general-events topic。
总结
解决 Docker Compose 中 Kafka 消息发送失败的问题，关键在于确保 Kafka 的监听器配置正确，生产者使用正确的 Broker 地址，以及容器之间的网络连接正常。通过仔细检查这些配置，可以有效地解决此类连接问题，确保应用程序能够成功地向 Kafka 主题发送消息。
