Hey, I'm trying to run pretty heavy Flink job using PyFlink and Flink 1.16.2 on Kubernetes (using Flink Kubernetes Operator 1.5). After some time I'm receiving next exception. Any idea what it can be?

2023-06-16 00:19:51,850 ERROR org.apache.beam.runners.fnexecution.control.FnApiControlClient [] - FnApiControlClient received an error.
org.apache.beam.vendor.grpc.v1p43p2.io.grpc.StatusRuntimeException: CANCELLED: client cancelled
	at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.Status.asRuntimeException(Status.java:526) ~[blob_p-934ed733171eb46b5e06f37d5f96d10cd381ed93-977991e704a3c43c3b629ef06bb93e3c:1.16.2]

...

2023-06-16 00:19:51,852 WARN  org.apache.beam.runners.fnexecution.logging.GrpcLoggingService [] - Logging client failed unexpectedly.
org.apache.beam.vendor.grpc.v1p43p2.io.grpc.StatusRuntimeException: CANCELLED: client cancelled
	at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.Status.asRuntimeException(Status.java:526) ~[blob_p-934ed733171eb46b5e06f37d5f96d10cd381ed93-977991e704a3c43c3b629ef06bb93e3c:1.16.2]
	at org.apache.beam.vendor.grpc.v1p43p2.io.grpc.stub.ServerCalls$StreamingServerCallHandler$StreamingServerCallListener.onCancel(ServerCalls.java:291) [blob_p-934ed733171eb46b5e06f37d5f96d10cd381ed93-977991e704a3c43c3b629ef06bb93e3c:1.16.2]

...

2023-06-16 00:19:51,850 ERROR org.apache.beam.sdk.fn.data.BeamFnDataGrpcMultiplexer        [] - Failed to handle for unknown endpoint
org.apache.beam.vendor.grpc.v1p43p2.io.grpc.StatusRuntimeException: CANCELLED: client cancelled

....

2023-06-16 00:19:51,861 ERROR org.apache.beam.runners.fnexecution.control.FnApiControlClient [] - FnApiControlClient closed, clearing outstanding requests {1=java.util.concurrent.CompletableFuture@73abe63d[Not completed, 1 dependents]}

Next issues claim that there are deserialization issue with record, but actual cause is:

Caused by: org.apache.beam.vendor.grpc.v1p43p2.io.grpc.StatusRuntimeException: CANCELLED: call already cancelled. Use ServerCallStreamObserver.setOnCancelHandler() to disable this exception

@Yaroslav BezruchenkoYou could try to change

python.fn-execution.bundle.size

(default 1000 ) to a smaller value, e.g. 100.

Decreased to 50, not helping Loading configuration property: python.fn-execution.bundle.size, 50

@Yaroslav Bezruchenko It may happen that the Python process crashes because of memory exceeds the memory limit of the container and got killed by the OS. So another direction you could try is to tune the memory. You could tune the following configurations:

taskmanager.memory.managed.fraction (default is 0.4, could increase it)
taskmanager.memory.managed.consumer-weights (default is OPERATOR:70,STATE_BACKEND:70,PYTHON:30, could increase the weight of Python)
taskmanager.memory.jvm-overhead.fraction (default is 0.1, could increase it)
taskmanager.memory.process.size (no default value, if you use this configuration, could increase it)