Kafka Akka Actor抛出Java堆空间错误：Pod内存未耗尽的原因排查

我在OpenShift上运行一个基于Akka的Kafka应用，该应用会偶尔抛出Java堆空间错误：

Uncaught error from thread [transparency_16_1_a_ecp_aggregator-akka.actor.default-dispatcher-4]: Java heap space, shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled for ActorSystem[transparency_16_1_a_ecp_aggregator]

java.lang.OutOfMemoryError: Java heap space at java.base/java.nio.HeapByteBuffer.(Unknown Source) at java.base/java.nio.ByteBuffer.allocate(Unknown Source) at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30) at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:113) at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:452) at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:402) at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:674) at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:576) at org.apache.kafka.common.network.Selector.poll(Selector.java:481) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:560) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:236) at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1306) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1242) at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1215) at KafkaGeneratingUnitsRepository.getGeneratingUnits(KafkaGeneratingUnitsRepository.java:84) at .GeneratingUnitsConsumerActor.consumeGeneratingUnits(GeneratingUnitsConsumerActor.java:65) at actor.GeneratingUnitsConsumerActor$$Lambda$633/0x00000001005d2440.apply(Unknown Source) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) at scala.PartialFunction.applyOrElse(PartialFunction.scala:214) at scala.PartialFunction.applyOrElse$(PartialFunction.scala:213) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:269) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:270) at akka.actor.Actor.aroundReceive(Actor.scala:537) at akka.actor.Actor.aroundReceive$(Actor.scala:535) at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580) at akka.actor.ActorCell.invoke(ActorCell.scala:548) at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270) at akka.dispatch.Mailbox.run(Mailbox.scala:231)

进而引发以下错误：

java.lang.OutOfMemoryError: Java heap space at java.base/java.nio.HeapByteBuffer.(Unknown Source) at java.base/java.nio.ByteBuffer.allocate(Unknown Source) at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30) at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:113) at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:452) at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:402) at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:674) at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:576) at org.apache.kafka.common.network.Selector.poll(Selector.java:481) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:560) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:306) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.maybeLeaveGroup(AbstractCoordinator.java:1114) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:1074) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:980) at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2411) at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2378) at KafkaGeneratingUnitsRepository.getGeneratingUnits(KafkaGeneratingUnitsRepository.java:92) at actor.GeneratingUnitsConsumerActor.consumeGeneratingUnits(GeneratingUnitsConsumerActor.java:65) at actor.GeneratingUnitsConsumerActor$$Lambda$633/0x00000001005d2440.apply(Unknown Source) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) at scala.PartialFunction.applyOrElse(PartialFunction.scala:214) at scala.PartialFunction.applyOrElse$(PartialFunction.scala:213) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:269) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:270) at akka.actor.Actor.aroundReceive(Actor.scala:537) at akka.actor.Actor.aroundReceive$(Actor.scala:535) at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580) at akka.actor.ActorCell.invoke(ActorCell.scala:548)

以及：

org.apache.kafka.common.KafkaException: Failed to close kafka consumer at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2430) at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2378) at repository.KafkaGeneratingUnitsRepository.getGeneratingUnits(KafkaGeneratingUnitsRepository.java:92) at actor.GeneratingUnitsConsumerActor.consumeGeneratingUnits(GeneratingUnitsConsumerActor.java:65) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) at scala.PartialFunction.applyOrElse(PartialFunction.scala:214) at scala.PartialFunction.applyOrElse$(PartialFunction.scala:213) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:269) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:270) at akka.actor.Actor.aroundReceive(Actor.scala:537) at akka.actor.Actor.aroundReceive$(Actor.scala:535) at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580) at akka.actor.ActorCell.invoke(ActorCell.scala:548) at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270) at akka.dispatch.Mailbox.run(Mailbox.scala:231) at akka.dispatch.Mailbox.exec(Mailbox.scala:243) at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source) at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown Source) at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source) at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source) at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source) Caused by: java.lang.OutOfMemoryError: Java heap space at java.base/java.nio.HeapByteBuffer.(Unknown Source) at java.base/java.nio.ByteBuffer.allocate(Unknown Source) at org.apache.kafka.common.memory.MemoryPool$1.tryAllocate(MemoryPool.java:30) at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:113) at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:452) at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:402) at org.apache.kafka.common.network.Selector.attemptRead(Selector.java:674) at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:576) at org.apache.kafka.common.network.Selector.poll(Selector.java:481) at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:560) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:265) at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.pollNoWakeup(ConsumerNetworkClient.java:306) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.maybeLeaveGroup(AbstractCoordinator.java:1114) at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.close(AbstractCoordinator.java:1074) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.close(ConsumerCoordinator.java:980) at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2411) at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2378) at repository.KafkaGeneratingUnitsRepository.getGeneratingUnits(KafkaGeneratingUnitsRepository.java:92) at actor.GeneratingUnitsConsumerActor.consumeGeneratingUnits(GeneratingUnitsConsumerActor.java:65) at actor.GeneratingUnitsConsumerActor$$Lambda$633/0x00000001005d2440.apply(Unknown Source) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24) at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20) at scala.PartialFunction.applyOrElse(PartialFunction.scala:214) at scala.PartialFunction.applyOrElse$(PartialFunction.scala:213) at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:269) at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:270) at akka.actor.Actor.aroundReceive(Actor.scala:537) at akka.actor.Actor.aroundReceive$(Actor.scala:535) at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220) at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580) at akka.actor.ActorCell.invoke(ActorCell.scala:548)

查看Pod的内存使用情况（），看起来并不应该触发堆空间错误，请问该错误的可能原因是什么？

1. 瞬时内存尖峰未被监控捕获 #

监控面板展示的通常是时间段内的平均内存值，而OOM错误往往由短时间内的内存突增导致——比如Kafka消费者一次性拉取大量大消息，内存占用瞬间超过堆上限，这种尖峰可能在监控采样间隔中被平均掉，无法在面板上体现。

2. Kafka消费者配置不合理 #

从栈轨迹看，错误发生在Kafka消费者拉取消息的网络IO阶段（NetworkReceive.readFrom），可能的配置问题包括：

• fetch.max.bytes/max.partition.fetch.bytes设置过大：消费者单次从Kafka拉取的消息总大小超过堆内存剩余空间；
• max.poll.records值过高：一次性拉取过多消息到内存中处理，导致堆内存耗尽；
• 内存池配置缺失：Kafka默认使用的内存池（MemoryPool$1）如果未合理配置，会频繁分配大ByteBuffer，加剧内存压力。

3. Akka Actor消息积压 #

Akka Actor的邮箱如果出现消息积压，会导致大量未处理的消息驻留在堆内存中。当Kafka消费者持续向Actor发送消息，而Actor处理速度跟不上时，邮箱中的消息会不断累积，最终引发堆内存不足。

4. 内存泄漏 #

• 消费者资源未正确释放：栈轨迹显示关闭Kafka消费者时也触发了OOM，说明消费者实例可能未被及时回收，或存在重复创建消费者但未关闭的情况，导致内存中堆积大量消费者相关对象；
• 业务对象内存泄漏：处理Kafka消息时生成的业务对象（如GeneratingUnits相关实例）未被正确回收，长期驻留在堆中（比如被静态集合引用、Actor状态未清理等）。

5. JVM堆内存与Pod资源不匹配 #

OpenShift中Pod的内存限制（resources.limits.memory）如果小于JVM堆内存（-Xmx）设置，会导致JVM尝试分配超过Pod可用的内存，触发OOM；反之，如果Pod内存限制远大于JVM堆内存，但JVM堆设置过小，也会在业务高峰时出现堆内存不足。需确认-Xmx、-Xms与Pod内存限制的配置比例是否合理（通常堆内存设置为Pod内存限制的50%-70%）。

6. 直接内存与堆内存的竞争 #

如果应用使用大量直接内存（Direct ByteBuffer），虽然直接内存不在堆中，但JVM总内存占用（堆+直接内存+元空间等）不能超过Pod的内存限制。当直接内存占用过高时，会挤压堆内存的可用空间，间接引发堆内存OOM。

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