confluentinc/cp-kafka KRaft yaml (zookeeper out)

https://fullmooney.tistory.com/75

confluentinc kafka, zookeeper, kafka-ui

위 링크에서 한 구성은 전통적인 방식의 zookeeper 클러스터에서 메타데이터를 관리하고 kafka클러스터를 그 위에서 동작하도록 하는 방식이다.

 

KRaft 설정을 통해 zookeeper를 제외한 상태로 k8s에 배포해 보자.

도커컴포즈는 이 블로그를 참고했다.

https://medium.com/mo-zza/kafka-kraft-%EB%AA%A8%EB%93%9C-with-docker-%EB%8F%99%EB%AC%BC%EC%9B%90%EC%9D%84-%ED%83%88%EC%B6%9C%ED%95%9C-kafka-8b5e7c7632fa 

Kafka KRaft 모드 (with Docker) - 동물원을 탈출한 Kafka

 

기존의 deployment를 Statefulset으로 변경하고,  configmap에 일부 환경변수 추가하였다.

 

먼저 configmap이다. 기존 zookeeper 관련 내용은 모두 삭제했다.

apiVersion: v1
data:
  CLUSTER_ID: ${KAFKA_CLUSTER_ID:-MkU3OEVBNTcwNTJENDM2Qk} # 신규
  KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
  KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka1-0:29092,kafka2-0:29092,kafka3-0:29092 # 변경. kafka-ui에서 사용
  KAFKA_CLUSTERS_0_NAME: test-cluster
  KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER #신규
  KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka1-0:29093,2@kafka2-0:29093,3@kafka3-0:29093 #신규
  KAFKA_DEFAULT_REPLICATION_FACTOR: "3"
  KAFKA_DELETE_TOPIC_ENABLE: "true"
  KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT #변경
  KAFKA_MIN_INSYNC_REPLICAS: "2"
  KAFKA_NUM_PARTITIONS: "1"
  KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: "3"
  KAFKA_PROCESS_ROLES: broker,controller # 신규
  KAFKA_UI_AUTH_ENABLED: "false"
kind: ConfigMap
metadata:
  annotations:
  name: kafka-cm
  namespace: default

 

그리고 kafka1, kafka2, kafka3의 statefulset과 kafka1-0, kafka2-0, kafka3-0의 서비스를 생성했다.

vote와 listener에서 사용할 29092, 29093포트를 추가로 서비스에 명시했다.

apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    app: kafka1-0
  name: kafka1
  namespace: default
spec:
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: kafka1-0
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: kafka1-0
    spec:
      containers:
      - env:
        - name: KAFKA_ADVERTISED_LISTENERS
          value: INTERNAL://:29092,EXTERNAL://kafka1-0:9092
        - name: KAFKA_LISTENERS
          value: INTERNAL://:29092,CONTROLLER://:29093,EXTERNAL://0.0.0.0:9092
        - name: KAFKA_BROKER_ID
          value: '1'
        envFrom:
        - configMapRef:
            name: kafka-cm
        image: confluentinc/cp-kafka:latest
        imagePullPolicy: Always
        name: kafka1
        ports:
        - containerPort: 9092
          protocol: TCP
        resources: {}
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30

---

apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka1-0
  name: kafka1-0
  namespace: default
spec:
  internalTrafficPolicy: Cluster
  ipFamilies:
  - IPv4
  ipFamilyPolicy: SingleStack
  ports:
  - port: 9092
    protocol: TCP
    targetPort: 9092
    name: kafka-port
  - port: 29092  #추가
    protocol: TCP
    targetPort: 29092
    name: listen
  - port: 29093 #추가
    protocol: TCP
    targetPort: 29093
    name: vote
  selector:
    app: kafka1-0
  sessionAffinity: None
  type: ClusterIP
  
---

apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    app: kafka2-0
  name: kafka2
  namespace: default
spec:
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: kafka2-0
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: kafka2-0
    spec:
      containers:
      - env:
        - name: KAFKA_ADVERTISED_LISTENERS
          value: INTERNAL://:29092,EXTERNAL://kafka2-0:9093
        - name: KAFKA_LISTENERS
          value: INTERNAL://:29092,CONTROLLER://:29093,EXTERNAL://0.0.0.0:9093
        - name: KAFKA_BROKER_ID
          value: '2'
        envFrom:
        - configMapRef:
            name: kafka-cm
        image: confluentinc/cp-kafka:latest
        imagePullPolicy: Always
        name: kafka2
        ports:
        - containerPort: 9093
          protocol: TCP
        resources: {}
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30

---

apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka2-0
  name: kafka2-0
  namespace: default
spec:
  internalTrafficPolicy: Cluster
  ipFamilies:
  - IPv4
  ipFamilyPolicy: SingleStack
  ports:
  - port: 9093
    protocol: TCP
    targetPort: 9093
    name: kafka-port
  - port: 29092
    protocol: TCP
    targetPort: 29092
    name: listen
  - port: 29093
    protocol: TCP
    targetPort: 29093
    name: vote
  selector:
    app: kafka2-0
  sessionAffinity: None
  type: ClusterIP
  
---

apiVersion: apps/v1
kind: StatefulSet
metadata:
  labels:
    app: kafka3-0
  name: kafka3
  namespace: default
spec:
  replicas: 1
  revisionHistoryLimit: 10
  selector:
    matchLabels:
      app: kafka3-0
  template:
    metadata:
      creationTimestamp: null
      labels:
        app: kafka3-0
    spec:
      containers:
      - env:
        - name: KAFKA_ADVERTISED_LISTENERS
          value: INTERNAL://:29092,EXTERNAL://kafka3-0:9094
        - name: KAFKA_LISTENERS
          value: INTERNAL://:29092,CONTROLLER://:29093,EXTERNAL://0.0.0.0:9094
        - name: KAFKA_BROKER_ID
          value: '3'
        envFrom:
        - configMapRef:
            name: kafka-cm
        image: confluentinc/cp-kafka:latest
        imagePullPolicy: Always
        name: kafka3
        ports:
        - containerPort: 9094
          protocol: TCP
        resources: {}
        terminationMessagePath: /dev/termination-log
        terminationMessagePolicy: File
      dnsPolicy: ClusterFirst
      restartPolicy: Always
      schedulerName: default-scheduler
      securityContext: {}
      terminationGracePeriodSeconds: 30

---

apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka3-0
  name: kafka3-0
  namespace: default
spec:
  internalTrafficPolicy: Cluster
  ipFamilies:
  - IPv4
  ipFamilyPolicy: SingleStack
  ports:
  - port: 9094
    protocol: TCP
    targetPort: 9094
    name: kafka-port
  - port: 29092
    protocol: TCP
    targetPort: 29092
    name: listen
  - port: 29093
    protocol: TCP
    targetPort: 29093
    name: vote
  selector:
    app: kafka3-0
  sessionAffinity: None
  type: ClusterIP

 

확인을 위해 kafka-ui 는 pod만 삭제하여 변경된 환경변수를 가져올 수 있도록 하였다.

최종적으로 배포된 파드는 다음과 같다.

(⎈|docker-desktop:N/A) ~$ kubectl get po,svc 
NAME                                                      READY   STATUS    RESTARTS       AGE
pod/kafka-ui-686745fd88-2rstr                             1/1     Running   0              18m
pod/kafka1-0                                              1/1     Running   0              18m
pod/kafka2-0                                              1/1     Running   0              18m
pod/kafka3-0                                              1/1     Running   0              18m

NAME                                          TYPE           CLUSTER-IP       EXTERNAL-IP   PORT(S)                                             AGE
service/kafka-ui                              NodePort       10.104.54.219    <none>        8080:32051/TCP,8081:31119/TCP                       47h
service/kafka1-0                              ClusterIP      10.104.169.118   <none>        9092/TCP,29092/TCP,29093/TCP                        24m
service/kafka2-0                              ClusterIP      10.97.51.60      <none>        9093/TCP,29092/TCP,29093/TCP                        22m
service/kafka3-0                              ClusterIP      10.101.234.0     <none>        9094/TCP,29092/TCP,29093/TCP                        21m

 

kafka-ui 에 노드포트로 접근해서 보자.

정상으로 잘 동작하고 있다.