Hi all! I'm a beginner trying to run a python Beam pipeline (batch) on a flink session cluster using the flinkrunner. I'm able to get my job running on a small input corpus, but am facing several issues when trying to scale to larger corpuses. For context, I'm hosting the session cluster on k8s. Each TM is has one task slot. The taskmanager memory configs are as follows:

INFO  [] - Final TaskExecutor Memory configuration:
INFO  [] -   Total Process Memory:          15.625gb (16777216000 bytes)
INFO  [] -     Total Flink Memory:          14.375gb (15435038720 bytes)
INFO  [] -       Total JVM Heap Memory:     4.031gb (4328521728 bytes)
INFO  [] -         Framework:               128.000mb (134217728 bytes)
INFO  [] -         Task:                    3.906gb (4194304000 bytes)
INFO  [] -       Total Off-heap Memory:     10.344gb (11106516992 bytes)
INFO  [] -         Managed:                 5.750gb (6174015580 bytes)
INFO  [] -         Total JVM Direct Memory: 4.594gb (4932501412 bytes)
INFO  [] -           Framework:             256.000mb (268435456 bytes)
INFO  [] -           Task:                  256.000mb (268435456 bytes)
INFO  [] -           Network:               4.094gb (4395630500 bytes)
INFO  [] -     JVM Metaspace:               256.000mb (268435456 bytes)
INFO  [] -     JVM Overhead:                1024.000mb (1073741824 bytes)

Questions: 1.

With my task manager running only one worker slot, does it make sense to allocate more than one 1 CPU to each taskmanager pod?

I notice that even if I allocate 4 CPU to a taskmanager,

kubectl top pods

only shows 1000mCPU being used max at any given time. If not, is there a way to scale compute on taskmanagers? I'm also not clear if the cpu being shown by

kubectl top pods

is from the TM or the beam worker pool, one path to investigate further is to instrument container level metrics, but would love to hear any expertise on this beforehand. • I have tried the

taskmanager.cpu.cores

and

kubernetes.taskmanager.cpu

flags, and these seem to be acked by the resource manager (from taking a look at jobmanager logs), but the actual utilization doesn't seem to match 2.

Is there a heuristic for tuning direct buffer memory size? How can I debug shuffle issues? I see the following errors after my input stage finishes reading (the flink web UI shows the stage is running and that records (32GB) have been sent, but it seems to hang here). I don't see any logs on my jobmanager or taskmanager outside the following exception:

Oct 06, 2023 7:34:50 PM org.apache.beam.vendor.grpc.v1p54p0.io.grpc.netty.NettyServerTransport notifyTerminated
INFO: Transport failed
java.lang.OutOfMemoryError: Direct buffer memory
	at java.base/java.nio.Bits.reserveMemory(Unknown Source)
	at java.base/java.nio.DirectByteBuffer.<init>(Unknown Source)
	at java.base/java.nio.ByteBuffer.allocateDirect(Unknown Source)

• 256mb of direct jvm task memory worked on a smaller corpus (3GB). For this larger corpus, I also tried scaling direct mem to 512mb, but it OOMed as well. Could someone help me understand how direct memory usage scales with network shuffle size? (I have read the taskmanager memory configuration docs but didn't find it granular enough) Would greatly appreciate any insight from the community, and please let me know if there's any more information I can provide that could help - thanks!