k8s Kafka Template

Kafka — ключевое решение для организаций, которые заинтересованы во внедрении событийно-управляемых архитектур и систем реального времени.

00-namespace.yaml

apiVersion: v1
kind: Namespace
metadata:
  name: "kafka"
  labels:
    name: "kafka"

kubectl apply -f 00-namespace.yaml

01-zookeeper.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    app: zookeeper-service
  name: zookeeper-service
  namespace: kafka
spec:
  type: NodePort
  ports:
    - name: zookeeper-port
      port: 2181
      nodePort: 30181
      targetPort: 2181
  selector:
    app: zookeeper
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: zookeeper
  name: zookeeper
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: zookeeper
  template:
    metadata:
      labels:
        app: zookeeper
    spec:
      containers:
        - image: wurstmeister/zookeeper
          imagePullPolicy: IfNotPresent
          name: zookeeper
          ports:
            - containerPort: 2181

kubectl apply -f 01-zookeeper.yaml

kubectl get services -n kafka

Мы видим внутренний IP-адрес Zookeeper, который нам понадобится сообщить брокеру, где его прослушивать.

Последний шаг — развертывание брокера Kafka. Мы создаем02-kafka.yamlфайл со следующим содержимым, заменяем его<ZOOKEEPER-INTERNAL-IP>наCLUSTER-IPпредыдущий шаг для Zookeeper. Если не выполнить этот шаг, брокер не сможет выполнить развертывание.

02-kafka.yaml

apiVersion: v1
kind: Service
metadata:
  labels:
    app: kafka-broker
  name: kafka-service
  namespace: kafka
spec:
  ports:
  - port: 9092
  selector:
    app: kafka-broker
---
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: kafka-broker
  name: kafka-broker
  namespace: kafka
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-broker
  template:
    metadata:
      labels:
        app: kafka-broker
    spec:
      hostname: kafka-broker
      containers:
      - env:
        - name: KAFKA_BROKER_ID
          value: "1"
        - name: KAFKA_ZOOKEEPER_CONNECT
          value: <ZOOKEEPER-INTERNAL-IP>:2181
        - name: KAFKA_LISTENERS
          value: PLAINTEXT://:9092
        - name: KAFKA_ADVERTISED_LISTENERS
          value: PLAINTEXT://kafka-broker:9092
        image: wurstmeister/kafka
        imagePullPolicy: IfNotPresent
        name: kafka-broker
        ports:
          - containerPort: 9092

kubectl apply -f 02-kafka.yaml

Просмотреть запущенные сервисы

kubectl get pods -n kafka

(pythonProject) PS J:\MTC-aquarium\k8s> kubectl get pods -n kafka NAME READY STATUS RESTARTS AGE kafka-broker-75d58df49b-tt69x 1/1 Running 0 4m18s zookeeper-74d96f84bd-cnjj2 1/1 Running 0 16m

Чтобы проверить, можем ли мы отправлять и получать сообщения из темы в Kafka, нам нужно будет открыть порт для Kafka, чтобы сделать его доступным с локального хоста. Мы запускаем следующую команду, чтобы открыть порт:

kubectl port-forward kafka-broker-75d58df49b-tt69x 9092 -n kafka

Эта команда делает порт9092этого модуля доступным за пределами кластера Minikube k8s по адресуlocalhost:9092.

Чтобы легко отправлять и получать сообщения из Kafka, мы будем использовать инструмент командной строкиKCat(ранее Kafkacat). Чтобы создать сообщение и тему с именемtest, мы запускаем следующую команду:

echo "hello world!" | kafkacat -P -b localhost:9092 -t test