Hello, I’m running pyflink jobs in session mode using the flink-kubernetes-operator 1.4.0 and flink 1.17. When I submit multiple jobs at the same time, some jobs do not start and other jobs will start multiple times. However, the total number of jobs is always correct. Logs from the kubernetes operator always show each of the distinct jobs in the running state, but the flink ui will show the jobs that are actually running. Ex, if we have job1, job2, and job3; we may see job1 running twice, job 2 running one, and job3 not running at all. When deployed one at a time, all jobs start as expected. Are there any configurations or other recommendations that could resolve this issue?

@Alex T Could you share more information, e.g. the yaml which defines the job, etc.

This is the helm template I use for the jobs.

metadata.name

and

spec.job.args

will change for each job.

{{- range .<http://Values.jobs|Values.jobs> }}
apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkSessionJob
metadata:
  name: {{ .name }}
spec:
  deploymentName: flink-session-deployment
  job:
    jarURI: <https://repo1.maven.org/maven2/org/apache/flink/flink-python/1.17.0/flink-python-1.17.0.jar> # Note, this jarURI is actually a placeholder
    entryClass: "org.apache.flink.client.python.PythonDriver"
    args: ["-py", {{ .file | quote }}]
    parallelism: 1
    upgradeMode: stateless
---
{{- end }}

This is the template I use for the FlinkDeployment

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkDeployment
metadata:
  name: flink-session-deployment
spec:
  image: "{{ $.Values.image.repository }}:{{ $.Values.image.tag }}"
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "16"
    rest.flamegraph.enabled: "false"
    high-availability: kubernetes
    metrics.reporter.prom.factory.class: org.apache.flink.metrics.prometheus.PrometheusReporterFactory
    metrics.reporter.prom.port: "9999"
    restart-strategy: exponential-delay
    restart-strategy.exponential-delay.initial-backoff: "60 s"
    restart-strategy.exponential-delay.max-backoff: "4 min"
    restart-strategy.exponential-delay.backoff-multiplier: "2.0"
    restart-strategy.exponential-delay.reset-backoff-threshold: "10 min"
    restart-strategy.exponential-delay.jitter-factor: "0.2"
    high-availability.storageDir: <s3://bucket>
  serviceAccount: flink
  jobManager:
    resource:
      memory: "8g"
      cpu: 1
  taskManager:
    resource:
      memory: "64000m"
      cpu: 16
  podTemplate:
    apiVersion: v1
    kind: Pod
    metadata:
      annotations:
        <http://prometheus.io/path|prometheus.io/path>: /
        <http://prometheus.io/port|prometheus.io/port>: "9999"
        <http://prometheus.io/scrape|prometheus.io/scrape>: "true"