已连接到组协调器但无法连接到 kafka 节点

Posted 2023-03-28

【中文标题】已连接到组协调器但无法连接到 kafka 节点

【英文标题】：Connected to a group coordinator but can't connect to kafka node

【发布时间】：2021-07-26 00:26:51

【问题描述】：

我正在尝试设置一个集成测试，其中我有一个 kafka 生产者（quarkus / 标准 kafkaproducer，dockerized）将内容推送到一个主题，然后我的应用程序（quarkus / 反应式消息传递，也是 dockerized）接收它并推送到另一个主题来通知它已收到消息（主题：消息接收）。在我的测试中，我有一个 KafkaContainer

@Container
static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.4"))
        .withNetworkAliases("kafkaNode")
        .withNetwork(CommonNetwork.getInstance())
        .withExposedPorts(9092, 9093);

我的两个 dockerized 应用都运行良好，因为它们都共享同一个网络，发送/接收/发送场景正在运行。但是我在实际测试中遇到了 KafkaConsumer 问题，他应该检查主题 messageReceived 以获取来自我的应用程序的消息。我正在使用以下内容将引导服务器配置添加到我的消费者，这似乎正在工作：

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());

它使用来自 testcontainers 的随机 ip 发现组协调器，因此配置必须正确。

Discovered group coordinator localhost:49725 (id: 2147483646 rack: null)

但随后它会尝试连接到以下内容：

Connection to node -1 (localhost/127.0.0.1:9092) could not be established. Broker may not be available.
Bootstrap broker localhost:9092 (id: -1 rack: null) disconnected

当它应该连接到与组协调器相同的 localhost:49725 时，因为如果我理解正确，我只有一个节点充当代理和组协调器。

我可能做错了什么，但我不知道是什么。可以在这件事上使用一些帮助。谢谢。

【参考方案1】：

我在探索 TestContainers 进行集成测试时遇到了类似的问题。我设法用下面的代码解决了它：

KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"));
        kafkaContainer.setPortBindings(Arrays.asList("9092:9092", "9093:9093", "2181:2181"));
        kafkaContainer.start();
        kafkaContainer.waitingFor(Wait.forListeningPort());

KafkaConsumer 使用以下方式连接：

props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "PLAINTEXT://localhost:9093");

这里的关键是PLAINTEXT://localhost:9093。

【讨论】：

感谢您的输入，这可能会解决我的问题，但我希望有一些不需要手动绑定端口的东西。根据我的阅读，这不是一个好的做法，并且可能会导致 CI​​ 环境出现问题，因为这些端口可能会被占用。

好的。所以你可以有自己的逻辑从这个kafka.getBootstrapServers()中取出端口，然后使用PLAINTEXT://localhost:<PORT>