apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: {{ .Release.Name }}-flink-operator
  labels:
    app: {{ .Release.Name }}-flink-operator
    component: {{ .Release.Name }}-flink-operator
spec:
  serviceName: {{ .Release.Name }}-flink-operator
  replicas: 1
  selector:
    matchLabels:
      app: {{ .Release.Name }}-flink-operator
      component: {{ .Release.Name }}-flink-operator
  template:
    metadata:
      labels:
        app: {{ .Release.Name }}-flink-operator
        component: {{ .Release.Name }}-flink-operator
    spec:
      serviceAccountName: {{ include "flink-kube-operator.serviceAccountName" . }}
      initContainers:
        - name: wait-for-jobmanager
          image: curlimages/curl:8.5.0  # Lightweight curl image
          command:
            - sh
            - -c
            - |
              echo "Waiting for Flink JobManager to be ready..."
              until curl -sSf "http://{{ .Release.Name }}-flink-job-manager:8081/taskmanagers"; do
                echo "JobManager not ready yet - retrying in 5s..."
                sleep 5
              done
              echo "JobManager is ready!"
      containers:
        - name: operator
          securityContext:
            {{- toYaml .Values.securityContext | nindent 12 }}
          image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
          imagePullPolicy: {{ .Values.image.pullPolicy }}
          ports:
            - name: http
              containerPort: {{ .Values.service.port }}
              protocol: TCP
          env:
            - name: FLINK_API_URL
              value: {{ .Release.Name }}-flink-job-manager:8081
            - name: NAMESPACE
              value: "{{ .Release.Namespace }}"
            {{- if eq .Values.flink.state.savepoint.storageType "s3" }}
            - name: SAVEPOINT_PATH
              value: s3://flink/savepoints/
            - name: S3_ENDPOINT
              value: "http://{{ .Release.Name }}-minio:9000"
            - name: AWS_ACCESS_KEY_ID
              valueFrom:
                secretKeyRef:
                  name: {{ .Release.Name }}-minio
                  key: root-user
            - name: AWS_SECRET_ACCESS_KEY
              valueFrom:
                secretKeyRef:
                  name: {{ .Release.Name }}-minio
                  key: root-password
            {{- else }}
            - name: SAVEPOINT_PATH
              value: /opt/flink/savepoints/
            {{- end }}