I recently introduced ZooKeeper to my Flink Job for Checkpointing and High Availability. Things were running smoothly until a ZooKeeper leadership changed and the JobManager disconnected. The pattern I noticed is that the JobManager continued to try to reconnect to ZK, and eventually did when the new leader was elected. However, ~1 minute later, both the Job Manager + Task Manager were experiencing OutOfMemoryError and the Flink job crashed.
<code>org.apache.flink.util.FlinkException: Disconnect from JobManager responsible for 00000000000000000000000000000000.
Caused by: java.lang.Exception: Job leader for job id 00000000000000000000000000000000 lost leadership.
<code>org.apache.flink.util.FlinkException: Disconnect from JobManager responsible for 00000000000000000000000000000000.
...
Caused by: java.lang.Exception: Job leader for job id 00000000000000000000000000000000 lost leadership.
</code>
org.apache.flink.util.FlinkException: Disconnect from JobManager responsible for 00000000000000000000000000000000.
...
Caused by: java.lang.Exception: Job leader for job id 00000000000000000000000000000000 lost leadership.
<code>[flink-dist-1.17.1.jar:1.17.1]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.lang.OutOfMemoryError: Direct buffer memory. The direct out-of-memory error has occurred. This can mean two things: either job(s) require(s) a larger size of JVM direct memory or there is a direct memory leak. The direct memory can be allocated by user code or some of its dependencies. In this case 'taskmanager.memory.task.off-heap.size' configuration option should be increased. Flink framework and its dependencies also consume the direct memory, mostly for network communication. The most of network memory is managed by Flink and should not result in out-of-memory error. In certain special cases, in particular for jobs with high parallelism, the framework may require more direct memory which is not managed by Flink. In this case 'taskmanager.memory.framework.off-heap.size' configuration option should be increased. If the error persists then there is probably a direct memory leak in user code or some of its dependencies which has to be investigated and fixed. The task executor has to be shutdown...
at java.nio.Bits.reserveMemory(Unknown Source) ~[?:?]
at java.nio.DirectByteBuffer.<init>(Unknown Source) ~[?:?]
at java.nio.ByteBuffer.allocateDirect(Unknown Source) ~[?:?]
at sun.nio.ch.Util.getTemporaryDirectBuffer(Unknown Source) ~[?:?]
at sun.nio.ch.IOUtil.read(Unknown Source) ~[?:?]
at sun.nio.ch.IOUtil.read(Unknown Source) ~[?:?]
at sun.nio.ch.SocketChannelImpl.read(Unknown Source) ~[?:?]
at org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:103)
<code>[flink-dist-1.17.1.jar:1.17.1]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.lang.OutOfMemoryError: Direct buffer memory. The direct out-of-memory error has occurred. This can mean two things: either job(s) require(s) a larger size of JVM direct memory or there is a direct memory leak. The direct memory can be allocated by user code or some of its dependencies. In this case 'taskmanager.memory.task.off-heap.size' configuration option should be increased. Flink framework and its dependencies also consume the direct memory, mostly for network communication. The most of network memory is managed by Flink and should not result in out-of-memory error. In certain special cases, in particular for jobs with high parallelism, the framework may require more direct memory which is not managed by Flink. In this case 'taskmanager.memory.framework.off-heap.size' configuration option should be increased. If the error persists then there is probably a direct memory leak in user code or some of its dependencies which has to be investigated and fixed. The task executor has to be shutdown...
at java.nio.Bits.reserveMemory(Unknown Source) ~[?:?]
at java.nio.DirectByteBuffer.<init>(Unknown Source) ~[?:?]
at java.nio.ByteBuffer.allocateDirect(Unknown Source) ~[?:?]
at sun.nio.ch.Util.getTemporaryDirectBuffer(Unknown Source) ~[?:?]
at sun.nio.ch.IOUtil.read(Unknown Source) ~[?:?]
at sun.nio.ch.IOUtil.read(Unknown Source) ~[?:?]
at sun.nio.ch.SocketChannelImpl.read(Unknown Source) ~[?:?]
at org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:103)
</code>
[flink-dist-1.17.1.jar:1.17.1]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: java.lang.OutOfMemoryError: Direct buffer memory. The direct out-of-memory error has occurred. This can mean two things: either job(s) require(s) a larger size of JVM direct memory or there is a direct memory leak. The direct memory can be allocated by user code or some of its dependencies. In this case 'taskmanager.memory.task.off-heap.size' configuration option should be increased. Flink framework and its dependencies also consume the direct memory, mostly for network communication. The most of network memory is managed by Flink and should not result in out-of-memory error. In certain special cases, in particular for jobs with high parallelism, the framework may require more direct memory which is not managed by Flink. In this case 'taskmanager.memory.framework.off-heap.size' configuration option should be increased. If the error persists then there is probably a direct memory leak in user code or some of its dependencies which has to be investigated and fixed. The task executor has to be shutdown...
at java.nio.Bits.reserveMemory(Unknown Source) ~[?:?]
at java.nio.DirectByteBuffer.<init>(Unknown Source) ~[?:?]
at java.nio.ByteBuffer.allocateDirect(Unknown Source) ~[?:?]
at sun.nio.ch.Util.getTemporaryDirectBuffer(Unknown Source) ~[?:?]
at sun.nio.ch.IOUtil.read(Unknown Source) ~[?:?]
at sun.nio.ch.IOUtil.read(Unknown Source) ~[?:?]
at sun.nio.ch.SocketChannelImpl.read(Unknown Source) ~[?:?]
at org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:103)
I can reproduce the OutOfMemoryError issue by manually trigger a Zookeeper leadership change. What I’m trying to understand is why would a Zookeeper leadership change result in OutOfMemoryError for the Direct buffer?