I am using Flink S3 sink and intermittently seeing following exception in S3 Sink operator which is triggering a cluster restart
Task - - Sink ConsumptionWriter (207/250)#0 (5a254b0622e8571f30567c39872691d1) switched from RUNNING to FAILED with failure cause: org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Lost connection to task manager 'ip-10-0-121-109.us-west-2.compute.internal/10.0.121.109:39219'. This indicates that the remote task manager was lost.
I have also seen following logs before lost connection exception, where I have noticed Total-Memory usage is more than available capacity. I am suspecting that could be the reason for cluster restart?
TaskManagerRunner - - Memory usage stats: [HEAP: 109905/206464/206464 MB, NON HEAP: 310/329/496 MB (used/committed/max)]
2022-09-01 17:08:08,422 INFO TaskManagerRunner - - Direct memory stats: Count: 1311709, Total Capacity: 43023060620, Used Memory: 43023060624
2022-09-01 17:08:08,422 INFO TaskManagerRunner - - Off-heap pool stats: [Code Cache: 123/124/240 MB (used/committed/max)], [Metaspace: 187/205/256 MB (used/committed/max)]
2022-09-01 17:08:08,422 INFO TaskManagerRunner - - Garbage collector stats: [G1 Young Generation, GC TIME (ms): 176063, GC COUNT: 2118], [G1 Old Generation, GC TIME (ms): 0, GC COUNT: 0]
TaskManagerRunner - - Memory usage stats: [HEAP: 111441/206464/206464 MB, NON HEAP: 310/329/496 MB (used/committed/max)]
2022-09-01 17:08:18,423 INFO TaskManagerRunner - - Direct memory stats: Count: 1311709, Total Capacity: 43023061644, Used Memory: 43023061648
2022-09-01 17:08:18,423 INFO TaskManagerRunner - - Off-heap pool stats: [Code Cache: 123/124/240 MB (used/committed/max)], [Metaspace: 187/205/256 MB (used/committed/max)]
2022-09-01 17:08:18,423 INFO TaskManagerRunner - - Garbage collector stats: [G1 Young Generation, GC TIME (ms): 176063, GC COUNT: 2118], [G1 Old Generation, GC TIME (ms): 0, GC COUNT: 0]
I am using r5d 24x large core node which has following configurations
r5d.24xlarge
96 vCore, 768 GiB memory, 3600 SSD GB storage
EBS Storage:none
Following is my flink configuration while submitting a job:
flink run \
-m yarn-cluster \
-d \
-yjm 491000 \
-ytm 485000 \
-yD taskmanager.numberOfTaskSlots=40 \
-yD jobmanager.io-pool.size=64 \
-yD jobmanager.future-pool.size=64 \
-yD jobmanager.memory.heap.size=460g \
-yD jobmanager.memory.jvm-overhead.max=40g \
-yD taskmanager.memory.process.size=440g \
-yD taskmanager.memory.jvm-overhead.max=50g \
-yD taskmanager.memory.network.max=50g \
-yD taskmanager.memory.network.min=40g \
-yD taskmanager.memory.managed.size=150g \
-yD taskmanager.memory.task.off-heap.size=4g \
-yD web.timeout=120000 \
-yD akka.ask.timeout=600s \
-yD heartbeat.timeout=600000 \
-yD rest.bind-port=8081 \
-yD akka.framesize="25006612b" \
-yD cluster.evenly-spread-out-slots=true \
-yD taskmanager.network.memory.buffer-debloat.enabled=true \
-yD state.checkpoints.dir=hdfs://ha-nn-uri/flink/checkpoints \
-yD state.savepoints.dir=hdfs://ha-nn-uri/flink/savepoints \
-yD high-availability.zookeeper.client.session-timeout=300000 \
-yD high-availability.zookeeper.client.connection-timeout=60000 \
-yD high-availability.zookeeper.client.max-retry-attempts=10 \
-yD high-availability.zookeeper.client.tolerate-suspended-connections=true \
-yD state.backend.local-recovery=true \
-yD taskmanager.debug.memory.log=true \
-yD taskmanager.debug.memory.log-interval=10000 \
-yD env.java.opts="-XX:+UnlockExperimentalVMOptions -Xloggc:/tmp/taskman.gc.log -XX:+PrintGCApplicationStoppedTime -XX:+PrintGCDetails -XX:+PrintGCDateStamps -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=10 -XX:GCLogFileSize=10M -XX:+PrintPromotionFailure -XX:+PrintGCCause -XX:+UseG1GC" \
$SAVEPOINT_CMD ATVPlaybackStateMachineFlinkJob-1.0-super-1.2.3.jar --stage "prod" --cell-name "StreamProcessor-Cell1-us-east-1"