apiVersion: v1
kind: Service
metadata: { name: kafka, namespace: broker }
spec:
  ports: [{ name: kafka, port: 9092, targetPort: 9092 }]
  selector: { app: kafka }
---
apiVersion: apps/v1
kind: StatefulSet
metadata: { name: kafka, namespace: broker }
spec:
  serviceName: kafka
  replicas: 1
  selector: { matchLabels: { app: kafka } }
  template:
    metadata: { labels: { app: kafka } }
    spec:
      nodeSelector: { node: hetzner-2 }
      containers:
      - name: kafka
        image: apache/kafka:latest
        env:
          - { name: KAFKA_NODE_ID, value: "1" }
          - { name: KAFKA_PROCESS_ROLES, value: "broker,controller" }
          - { name: KAFKA_LISTENERS, value: "PLAINTEXT://:9092,CONTROLLER://:9093" }
          - { name: KAFKA_ADVERTISED_LISTENERS, value: "PLAINTEXT://kafka.broker.svc.cluster.local:9092" }
          - { name: KAFKA_CONTROLLER_LISTENER_NAMES, value: "CONTROLLER" }
          - { name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP, value: "CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT" }
          - { name: KAFKA_CONTROLLER_QUORUM_VOTERS, value: "1@localhost:9093" }
          - { name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR, value: "1" }
          - { name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR, value: "1" }
          - { name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR, value: "1" }
          - { name: KAFKA_LOG_DIRS, value: "/var/lib/kafka/data" }
          - { name: CLUSTER_ID, value: "MkU3OEVBNTcwNTJENDM2Qk" }
        ports:
          - { containerPort: 9092 }
          - { containerPort: 9093 }
        volumeMounts:
          - { name: data, mountPath: /var/lib/kafka/data }
  volumeClaimTemplates:
  - metadata: { name: data }
    spec:
      accessModes: ["ReadWriteOnce"]
      storageClassName: local-ssd-hetzner
      resources: { requests: { storage: 50Gi } }