OpenShift内外连接EmbeddedKafkaBroker的问题与端口选择困惑

我在OpenShift服务器的容器中部署了EmbeddedKafkaBroker，配置如下：

EmbeddedKafkaBroker embeddedKafkaBroker = new EmbeddedKafkaBroker(1, false, 2);
embeddedKafkaBroker.kafkaPorts(8092);
embeddedKafkaBroker.addTopics("testTopic");
embeddedKafkaBroker.brokerProperty("listeners", "INTERNAL://0.0.0.0:8092,EXTERNAL://0.0.0.0:9092");
embeddedKafkaBroker.brokerProperty("advertised.listeners", "INTERNAL://localhost:8092,EXTERNAL://kafka-mock-daemon2017.apps.testing.soul:80");
embeddedKafkaBroker.brokerProperty("listener.security.protocol.map", "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT");
embeddedKafkaBroker.brokerProperty("inter.broker.listener.name", "INTERNAL");
embeddedKafkaBroker.afterPropertiesSet();
log.info("EmbeddedKafkaBroker started!");

容器内部端口9092映射到外部端口80，使用的依赖版本：

org.springframework.kafka:spring-kafka:2.7.10
org.springframework.kafka:spring-kafka-test:2.7.10

遇到的问题：

• 同一OpenShift命名空间内的其他应用连接kafka-mock-daemon2017.apps.testing.soul:80生产消息时，出现Java堆内存溢出错误
• 本地计算机应用连接该地址消费消息时，同样出现堆内存溢出错误

请问：

1. 配置哪里出错了？
2. OpenShift内部和本地计算机连接时，应使用INTERNAL还是EXTERNAL的host:port？

一、配置错误点 #

1. INTERNAL监听的对外公告地址错误
   OpenShift内部的每个容器都有独立网络命名空间，localhost:8092指向应用自身容器而非你的Kafka容器，内部应用无法通过该地址建立有效连接，会引发循环重试，最终导致堆内存溢出。

2. 端口配置冲突
   你同时调用了kafkaPorts(8092)和手动配置双listener端口（8092、9092），这会干扰EmbeddedKafkaBroker的端口初始化逻辑，可能导致部分端口绑定异常，客户端连接时出现资源耗尽问题。

3. 连接路径不匹配
   内部应用错误使用了EXTERNAL的对外地址，而该地址的网络转发路径针对外部流量设计，内部访问时会出现路由异常，引发客户端无限重试，最终触发OOM。

二、正确连接方式 #

• OpenShift内部同命名空间应用：使用INTERNAL listener的集群内部地址，格式为[服务名].[命名空间].svc.cluster.local:8092（比如你的服务名为kafka-mock-daemon2017，命名空间为testing，则地址是kafka-mock-daemon2017.testing.svc.cluster.local:8092）。
• 本地计算机应用：使用EXTERNAL listener的地址kafka-mock-daemon2017.apps.testing.soul:80，前提是OpenShift路由已正确将外部80端口转发到容器的9092端口。

三、修正后的配置示例 #

EmbeddedKafkaBroker embeddedKafkaBroker = new EmbeddedKafkaBroker(1, false, 2);
// 移除kafkaPorts的显式设置，让Broker绑定配置的双listener端口
embeddedKafkaBroker.addTopics("testTopic");
embeddedKafkaBroker.brokerProperty("listeners", "INTERNAL://0.0.0.0:8092,EXTERNAL://0.0.0.0:9092");
// 替换INTERNAL的对外公告地址为集群内部可访问的服务地址
embeddedKafkaBroker.brokerProperty("advertised.listeners", "INTERNAL://kafka-mock-daemon2017.testing.svc.cluster.local:8092,EXTERNAL://kafka-mock-daemon2017.apps.testing.soul:80");
embeddedKafkaBroker.brokerProperty("listener.security.protocol.map", "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT");
embeddedKafkaBroker.brokerProperty("inter.broker.listener.name", "INTERNAL");
embeddedKafkaBroker.afterPropertiesSet();
log.info("EmbeddedKafkaBroker started!");

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