Re: Flink k8s operator unstable deployment

[Naci Simsek]

Hi Arthur, In your initial mail, it was seen an explicit job id set: $internal.pipeline.job-id, 044d28b712536c1d1feed3475f2b8111 This might be the reason of duplicatedJobSubmission exception. In the job config on your last reply, I could not see such setting. You could verify from the JM logs that when the job needs to get resubmitted due to job failure, a randomized job-id must be defined by the operator. About HA, you are right, it is benefitial to use it for production in order for job to keep track of the checkpoints and help to recover from last checkpoint automatically. It is maybe not clearly stated in Flink doc for Kubernetes HA but in K8s HA case, there is always only 1 JM pod exists, there is NO standby replicated JM pods created. HA only works here as a job store, which this information related to the latest snapshot is getting passed to the newly created JM pod if existing one dies, to recover the job automatically. Regards, Naci On 30. Aug 2024, at 15:59, Arthur Catrisse <arthur.catri...@qonto.com> wrote: kubernetes.jobmanager.replicas, 1 execution.submit-failed-job-on-application-error, true high-availability.cluster-id, my_name kubernetes.jobmanager.cpu.limit-factor, 10 pipeline.max-parallelism, 6 kubernetes.service-account, flink kubernetes.cluster-id, my_name high-availability.storageDir, s3://my_bucket/recovery taskmanager.memory.flink.size, 1024m parallelism.default, 1 kubernetes.namespace, flink fs.s3a.aws.credentials.provider, com.amazonaws.auth.DefaultAWSCredentialsProviderChain kubernetes.jobmanager.owner.reference, apiVersion:flink.apache.org/v1beta1,kind:FlinkDeployment,uid:a42c1e37-8f5e-4ec0-a04f-00000,name:my_name,controller:false,blockOwnerDeletion:true state.backend.type, rocksdb kubernetes.container.image.ref, 00000.dkr.ecr.eu-west-3.amazonaws.com/data/my_image_ref jobmanager.memory.flink.size, 1024m taskmanager.memory.process.size, 2048m kubernetes.internal.jobmanager.entrypoint.class, org.apache.flink.kubernetes.entrypoint.KubernetesApplicationClusterEntrypoint pipeline.name, my_name execution.savepoint.path, s3://my_bucket/savepoints/savepoint-044d28-f5000c2e4bc0 state.backend.local-recovery, false state.backend.rocksdb.localdir, /opt/flink/state kubernetes.pod-template-file.taskmanager, /tmp/flink_op_generated_podTemplate_9407366845247969567.yaml state.backend.incremental, true web.cancel.enable, false execution.shutdown-on-application-finish, false job-result-store.delete-on-commit, false $internal.pipeline.job-id, 044d28b712536c1d1feed3475f2b8111 taskmanager.memory.managed.fraction, 0.6 $internal.flink.version, v1_19 execution.checkpointing.max-concurrent-checkpoints, 1 kubernetes.pod-template-file.jobmanager, /tmp/flink_op_generated_podTemplate_834737432685891333.yaml blob.server.port, 6102 kubernetes.jobmanager.annotations, flinkdeployment.flink.apache.org/generation:5 job-result-store.storage-path, s3://my_bucket/recovery/job-result-store/my_name/9cf5a2e7-89c6-40e7-94dd-c272a2007000 fs.allowed-fallback-filesystems, s3 high-availability.type, kubernetes state.savepoints.dir, s3://my_bucket/savepoints $internal.application.program-args, -pyclientexec;/usr/bin/python3;-py;/opt/flink/usrlib/my_name.py;--restoreMode;CLAIM taskmanager.numberOfTaskSlots, 2 kubernetes.rest-service.exposed.type, ClusterIP high-availability.jobmanager.port, 6101 process.working-dir, /tmp/workdir $internal.application.main, org.apache.flink.client.python.PythonDriver execution.target, kubernetes-application jobmanager.memory.process.size, 2048m taskmanager.rpc.port, 6100 internal.cluster.execution-mode, NORMAL kubernetes.jobmanager.tolerations, key:dedicated,operator:Equal,value:low-churn,effect:NoSchedule execution.checkpointing.externalized-checkpoint-retention, RETAIN_ON_CANCELLATION pipeline.jars, local:///opt/flink/opt/flink-python-1.19.0.jar state.checkpoints.dir, s3://my_bucket/checkpoints jobmanager.memory.off-heap.size, 134217728b jobmanager.memory.jvm-overhead.min, 805306368b jobmanager.memory.jvm-metaspace.size, 268435456b jobmanager.memory.heap.size, 939524096b jobmanager.memory.jvm-overhead.max, 805306368b Example of logs when failing Last pod before error : INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - RECEIVED SIGNAL 15: SIGTERM. Shutting down as requested. INFO org.apache.flink.runtime.entrypoint.ClusterEntrypoint [] - Shutting KubernetesApplicationClusterEntrypoint down with application status UNKNOWN. Diagnostics Cluster entrypoint has been closed externally.. (nothing else after, while other times, when no fail, we have among others terminating logs e.g org.apache.pekko.remote.RemoteActorRefProvider$RemotingTerminator [] - Shutting down remote daemon. etc) DuplicateJobSubmission Error crashloop : 2024-08-26 09:16:36,385 INFO org.apache.flink.client.deployment.application.executors.EmbeddedExecutor [] - Job e75c26c4679eb40acd89ed064665f9bb is submitted. 2024-08-26 09:16:36,385 INFO org.apache.flink.client.deployment.application.executors.EmbeddedExecutor [] - Submitting Job with JobId=e75c26c4679eb40acd89ed064665f9bb. 2024-08-26 09:16:36,510 INFO org.apache.flink.runtime.blob.FileSystemBlobStore [] - Creating highly available BLOB storage directory at s3://my_bucket/recovery/my-job-name/blob 2024-08-26 09:16:36,603 WARN org.apache.hadoop.fs.s3a.S3ABlockOutputStream [] - Application invoked the Syncable API against stream writing to recovery/my-job-name/blob/job_e75c26c4679eb40acd89ed064665f9bb/blob_p-dfdbbf7b0120ec73a562e6841144da852ede93ef-4db3fad9c1a552ff3e784386310bf022. This is unsupported 2024-08-26 09:16:36,614 WARN com.amazonaws.services.s3.internal.Mimetypes [] - Unable to find 'mime.types' file in classpath 2024-08-26 09:16:37,178 INFO org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Received JobGraph submission 'my-job-name' (e75c26c4679eb40acd89ed064665f9bb). 2024-08-26 09:16:37,299 WARN org.apache.flink.runtime.dispatcher.StandaloneDispatcher [] - Ignoring JobGraph submission 'my-job-name' (e75c26c4679eb40acd89ed064665f9bb) because the job already reached a globally-terminal state (i.e. FAILED, CANCELED, FINISHED) in a previous execution. Traceback (most recent call last): 2024-08-26 09:16:37,304 INFO org.apache.flink.client.python.PythonDriver [] - Traceback (most recent call last): File "/opt/flink/usrlib/my_job_name.py", line 159, in <module> 2024-08-26 09:16:37,304 INFO org.apache.flink.client.python.PythonDriver [] - File "/opt/flink/usrlib/my_job_name.py", line 159, in <module> publish_my_job_name() 2024-08-26 09:16:37,304 INFO org.apache.flink.client.python.PythonDriver [] - publish_my_job_name() File "/opt/flink/usrlib/my_job_name.py", line 127, in publish_my_job_name 2024-08-26 09:16:37,304 INFO org.apache.flink.client.python.PythonDriver [] - File "/opt/flink/usrlib/my_job_name.py", line 127, in publish_my_job_name table_env.execute_sql(""" 2024-08-26 09:16:37,304 INFO org.apache.flink.client.python.PythonDriver [] - table_env.execute_sql(""" File "/opt/flink/opt/python/pyflink.zip/pyflink/table/table_environment.py", line 837, in execute_sql 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - File "/opt/flink/opt/python/pyflink.zip/pyflink/table/table_environment.py", line 837, in execute_sql File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__ 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - File "/opt/flink/opt/python/py4j-0.10.9.7-src.zip/py4j/java_gateway.py", line 1322, in __call__ File "/opt/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 158, in deco 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - File "/opt/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 158, in deco pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Failed to execute sql 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - pyflink.util.exceptions.TableException: org.apache.flink.table.api.TableException: Failed to execute sql at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1060) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1060) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:876) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:876) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1112) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1112) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:735) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:735) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) at java.base/java.lang.reflect.Method.invoke(Unknown Source) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at java.base/java.lang.reflect.Method.invoke(Unknown Source) at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244) at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:374) at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282) at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132) at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79) at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238) at java.base/java.lang.Thread.run(Unknown Source) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at java.base/java.lang.Thread.run(Unknown Source) Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'my-job-name'. 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - Caused by: org.apache.flink.util.FlinkException: Failed to execute job 'my-job-name'. at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2455) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2455) at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:188) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:188) at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:110) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:110) at org.apache.flink.table.executor.python.ChainingOptimizingExecutor.executeAsync(ChainingOptimizingExecutor.java:88) 2024-08-26 09:16:37,305 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.table.executor.python.ChainingOptimizingExecutor.executeAsync(ChainingOptimizingExecutor.java:88) at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1032) 2024-08-26 09:16:37,306 INFO org.apache.flink.client.python.PythonDriver [] - at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1032) ... 14 more 2024-08-26 09:16:37,306 INFO org.apache.flink.client.python.PythonDriver [] - ... 14 more Caused by: org.apache.flink.runtime.client.JobSubmissionException: Job has already been submitted. 2024-08-26 09:16:37,306 INFO org.apache.flink.client.python.PythonDriver [] - Caused by: org.apache.flink.runtime.client.DuplicateJobSubmissionException: Job has already been submitted. at org.apache.flink.runtime.client.DuplicateJobSubmissionException.ofGloballyTerminated(DuplicateJobSubmissionException.java:35)