kafka 설정

2025. 2. 28. 16:42k8s/Kafka

StoragClass

apiVersion: storage.k8s.io/v1
kind: StorageClass
metadata:
  name: kafka-storage
provisioner: nfs.csi.k8s.io  # Use the correct NFS CSI driver
volumeBindingMode: Immediate

PV: statefulset의 replicas에 맞추처 만들면 됨

apiVersion: v1
kind: PersistentVolume
metadata:
  name: kafka-pv-0
spec:
  capacity:
    storage: 5Gi
  accessModes:
    - ReadWriteOnce
  storageClassName: kafka-storage  # Changed from nfs-storage to zookeeper-storage
  persistentVolumeReclaimPolicy: Retain
  claimRef:
    namespace: kafka
    name: data-kafka-0  # Ensures only this PVC binds to this PV
  nfs:
    path: {디렉토리}a/kafka-0
    server: {IP}

ConfigMap

apiVersion: v1
kind: ConfigMap
metadata:
  name: kafka-config
  namespace: kafka
data:
  KAFKA_ZOOKEEPER_CONNECT: "zookeeper-0.zookeeper-headless:2181,zookeeper-1.zookeeper-headless:2181,zookeeper-2.zookeeper-headless:2181"
  KAFKA_LISTENERS: "PLAINTEXT://:9092"
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT"
  KAFKA_INTER_BROKER_LISTENER_NAME: "PLAINTEXT"
  KAFKA_LOG_DIRS: "/var/lib/kafka/data"
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "3"
  KAFKA_DEFAULT_REPLICATION_FACTOR: "3"
  KAFKA_MIN_INSYNC_REPLICAS: "2"

StatefulSet

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: kafka
  namespace: kafka
spec:
  serviceName: kafka-headless
  replicas: 3  # ✅ Three Kafka brokers
  selector:
    matchLabels:
      app: kafka
  template:
    metadata:
      labels:
        app: kafka
    spec:
      terminationGracePeriodSeconds: 30
      containers:
        - name: kafka
          image: confluentinc/cp-kafka:{tag}
          ports:
            - containerPort: 9092
              name: client
          envFrom:
            - configMapRef:
                name: kafka-config
          command:
            - "/bin/bash"
            - "-c"
            - |
              export KAFKA_BROKER_ID=$((${HOSTNAME##*-} + 1))
              export KAFKA_ADVERTISED_LISTENERS="PLAINTEXT://$(hostname -f):9092"
              echo "Starting Kafka with Broker ID: $KAFKA_BROKER_ID"
              echo "Kafka Advertised Listeners: $KAFKA_ADVERTISED_LISTENERS"
              exec /etc/confluent/docker/run
          volumeMounts:
            - name: data
              mountPath: /var/lib/kafka/data
  volumeClaimTemplates:
    - metadata:
        name: data
      spec:
        accessModes:
          - ReadWriteOnce
        storageClassName: kafka-storage
        resources:
          requests:
            storage: 5Gi

Service

apiVersion: v1
kind: Service
metadata:
  name: kafka-headless
  namespace: kafka
spec:
  clusterIP: None  # ✅ Required for StatefulSet
  ports:
    - name: client
      port: 9092
      targetPort: 9092
  selector:
    app: kafka