# Изменено: 2026-04-06 (добавлен postStart hook для предсоздания топика iot.telemetry)
# kafka.yaml — минимальный деплой Apache Kafka в KRaft mode (без Zookeeper).
# Образ: apache/kafka (официальный, бесплатный).
# Используется для IoT telemetry pipeline: mqtt-bridge → Kafka → iot-kafka-consumer → Postgres.
# Для prod: заменить на managed Kafka (Confluent/Aiven) — только изменить KAFKA_BROKERS в Secret.
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-config
  namespace: sless
data:
  # server.properties для KRaft mode (без Zookeeper).
  # Нода совмещает роли controller + broker.
  server.properties: |
    process.roles=broker,controller
    node.id=1
    controller.quorum.voters=1@localhost:9093
    listeners=PLAINTEXT://:9092,CONTROLLER://:9093
    inter.broker.listener.name=PLAINTEXT
    controller.listener.names=CONTROLLER
    listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
    advertised.listeners=PLAINTEXT://kafka.sless.svc.cluster.local:9092
    log.dirs=/var/kafka-data/logs
    num.partitions=1
    default.replication.factor=1
    offsets.topic.replication.factor=1
    transaction.state.log.replication.factor=1
    transaction.state.log.min.isr=1
    log.retention.hours=168
    log.retention.check.interval.ms=300000
    auto.create.topics.enable=true
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  namespace: sless
  labels:
    app: kafka
spec:
  serviceName: kafka-headless
  replicas: 1
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      # apache/kafka образ запускается как UID 1000 (kafka user).
      # fsGroup=1000 — позволяет писать в PVC смонтированный как root.
      securityContext:
        fsGroup: 1000
      initContainers:
        # Форматирует хранилище KRaft если ещё не отформатировано.
        # KAFKA_CLUSTER_ID должен быть уникальным UUID — генерируется один раз.
        - name: kafka-init
          image: apache/kafka:3.7.0
          command:
            - /bin/sh
            - -c
            - |
              if [ ! -f /var/kafka-data/logs/meta.properties ]; then
                echo "Formatting Kafka storage..."
                /opt/kafka/bin/kafka-storage.sh format \
                  -t "$(cat /var/kafka-data/cluster.id 2>/dev/null || \
                        /opt/kafka/bin/kafka-storage.sh random-uuid | tee /var/kafka-data/cluster.id)" \
                  -c /tmp/kafka-config/server.properties
              fi
          volumeMounts:
            - name: kafka-data
              mountPath: /var/kafka-data
            - name: kafka-config
              mountPath: /tmp/kafka-config
      containers:
        - name: kafka
          image: apache/kafka:3.7.0
          command:
            - /opt/kafka/bin/kafka-server-start.sh
            - /tmp/kafka-config/server.properties
          ports:
            - containerPort: 9092
              name: client
            - containerPort: 9093
              name: controller
          resources:
            requests:
              cpu: 100m
              memory: 256Mi
            limits:
              cpu: 500m
              memory: 512Mi
          volumeMounts:
            - name: kafka-data
              mountPath: /var/kafka-data
            - name: kafka-config
              mountPath: /tmp/kafka-config
          readinessProbe:
            tcpSocket:
              port: 9092
            initialDelaySeconds: 30
            periodSeconds: 10
            failureThreshold: 6
      volumes:
        - name: kafka-config
          configMap:
            name: kafka-config
  volumeClaimTemplates:
    - metadata:
        name: kafka-data
      spec:
        accessModes: [ReadWriteOnce]
        storageClassName: vcd-disk-ext4
        resources:
          requests:
            storage: 1Gi
---
apiVersion: v1
kind: Service
metadata:
  name: kafka
  namespace: sless
  labels:
    app: kafka
spec:
  ports:
    - name: client
      port: 9092
      targetPort: 9092
  selector:
    app: kafka
---
apiVersion: v1
kind: Service
metadata:
  name: kafka-headless
  namespace: sless
  labels:
    app: kafka
spec:
  clusterIP: None
  ports:
    - name: client
      port: 9092
    - name: controller
      port: 9093
  selector:
    app: kafka