apiVersion: batch/v1
kind: Job
metadata:
  name: flink-jobmanager
spec:
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      restartPolicy: OnFailure
      containers:
        - name: jobmanager
          image: nicobiernat/theodolite-flink-uc3
          env:
            - name: KAFKA_BOOTSTRAP_SERVERS
              value: "my-confluent-cp-kafka:9092"
            - name: COMMIT_INTERVAL_MS
              value: "{{KAFKA_STREAMS_COMMIT_INTERVAL_MS}}"
            - name: JOB_MANAGER_RPC_ADDRESS
              value: "flink-jobmanager"
            - name: CHECKPOINTING
              value: "true"
            - name: FLINK_STATE_BACKEND
              value: "rocksdb"
            - name: FLINK_STATE_BACKEND_PATH
              value: "file:///opt/flink/statebackend"
            - name: FLINK_PROPERTIES
              value: |+
                blob.server.port: 6124
                jobmanager.rpc.port: 6123
                taskmanager.rpc.port: 6122
                queryable-state.proxy.ports: 6125
                jobmanager.memory.process.size: {{MEMORY_LIMIT}}
                taskmanager.memory.process.size: {{MEMORY_LIMIT}}
                parallelism.default: {{PARALLELISM}}
          resources:
            limits:
              memory: "{{MEMORY_LIMIT}}"
              cpu: "{{CPU_LIMIT}}"
          args: ["standalone-job", "--job-classname", "theodolite.uc3.application.HistoryServiceFlinkJob"] # optional arguments: ["--job-id", "<job id>", "--fromSavepoint", "/path/to/savepoint", "--allowNonRestoredState"]
          ports:
            - containerPort: 6123
              name: rpc
            - containerPort: 6124
              name: blob-server
            - containerPort: 8081
              name: webui
            - containerPort: 9249
              name: metrics
          livenessProbe:
            tcpSocket:
              port: 6123
            initialDelaySeconds: 30
            periodSeconds: 60
          volumeMounts:
            - name: flink-config-volume-rw
              mountPath: /opt/flink/conf
          securityContext:
            runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary
      initContainers:
        - name: init-jobmanager
          image: busybox:1.28
          command: ['cp', '-a', '/flink-config/.', '/flink-config-rw/']
          volumeMounts:
            - name: flink-config-volume
              mountPath: /flink-config/
            - name: flink-config-volume-rw
              mountPath: /flink-config-rw/
      volumes:
        - name: flink-config-volume
          configMap:
            name: flink-config
            items:
              - key: flink-conf.yaml
                path: flink-conf.yaml
              - key: log4j-console.properties
                path: log4j-console.properties
        - name: flink-config-volume-rw
          emptyDir: {}
