我们在Kubernetes上运行一个5节点的Flink集群(版本为1.6.3),使用了5个分区的Kafka主题作为数据源。有5个作业从该主题中读取数据(使用不同的消费者组),每个作业的并行度都为5。
每个任务管理器占用10GB的内存,任务管理器堆大小被限制为2GB。摄入负载较小(每秒100-200条消息),平均消息大小约为4-8KB。所有作业在几个小时内都可以正常运行。但在一段时间后,我们突然看到一个或多个作业失败了,原因是:
Flink在重启作业后仍然因为异常而失败。我们尝试了减少记录轮询(如此处建议的:Kafka消费者抛出java.lang.OutOfMemoryError:直接缓冲区内存),以及增加Kafka堆大小(如此处建议的:Flink + Kafka,当并行度> 1时java.lang.OutOfMemoryError)。但我不明白在flink进程中无法分配内存与kafka代理进程的jvm内存有什么关系,并且我没有看到任何指示oom的代理日志。请问还可能是什么原因导致失败?我们还应该检查什么?谢谢!
每个任务管理器占用10GB的内存,任务管理器堆大小被限制为2GB。摄入负载较小(每秒100-200条消息),平均消息大小约为4-8KB。所有作业在几个小时内都可以正常运行。但在一段时间后,我们突然看到一个或多个作业失败了,原因是:
ava.lang.OutOfMemoryError: Direct buffer memory
at java.nio.Bits.reserveMemory(Bits.java:666)
at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123)
at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:311)
at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:241)
at sun.nio.ch.IOUtil.read(IOUtil.java:195)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
at org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
at org.apache.kafka.common.network.NetworkReceive.readFromReadableChannel(NetworkReceive.java:97)
at org.apache.kafka.common.network.NetworkReceive.readFrom(NetworkReceive.java:71)
at org.apache.kafka.common.network.KafkaChannel.receive(KafkaChannel.java:169)
at org.apache.kafka.common.network.KafkaChannel.read(KafkaChannel.java:150)
at org.apache.kafka.common.network.Selector.pollSelectionKeys(Selector.java:355)
at org.apache.kafka.common.network.Selector.poll(Selector.java:303)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:349)
at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:226)
at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:1047)
at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:995)
at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:257)
Flink在重启作业后仍然因为异常而失败。我们尝试了减少记录轮询(如此处建议的:Kafka消费者抛出java.lang.OutOfMemoryError:直接缓冲区内存),以及增加Kafka堆大小(如此处建议的:Flink + Kafka,当并行度> 1时java.lang.OutOfMemoryError)。但我不明白在flink进程中无法分配内存与kafka代理进程的jvm内存有什么关系,并且我没有看到任何指示oom的代理日志。请问还可能是什么原因导致失败?我们还应该检查什么?谢谢!