云和安全管理服务专家新钛云服 祝祥 翻译

在本文中，我们将了解 Kafka 的架构以及它如何通过复制分区支持高可用性。然后，我们也可以自己实现一个 Kafka 集群以使用标准 Kubernetes 资源实现高可用性，并了解它是如何做到高可用的，从而避免单点故障问题。

在最简单的基础环境中，Kafka 的架构由单个 Broker 服务器及其作为客户端的生产者和消费者组成。

· 生产者创建记录并将它们发布到 Kafka 代理。

· 消费者从broker节点消费记录。

虽然这个 Kafka 集群可以支持典型的 Kafka 用例，但对于大多数实际情况来说它还是过于简单了。

首先，让我们解释一下常见的相关专业术语。

· Broker：消息中间件处理节点；每个Kafka服务节点称之为一个Broker，一个Kafka集群由一个或多个Broker组成。

· Topic：一类特定数据集合的统称；可类比DB中Table的概念；逻辑概念。

· Producer：消息的生产者，向Broker发送消息的客户端。

· Consumer：消息的消费者，向Broker读取消息的客户端。

· Consumer Group：每一个Consumer隶属于一个特定的Consumer Group，一条消息可以被不同Group中的Consumer消费，但同一Group内的消息只能被一个Consumer消费。

· Partition：是对Topic中所包含数据集的物理分区；物理概念。

· Replication：副本集；是Kafka高可用的一种保障机制。

Kafka 通常作为一个由三个或更多brokers代理组成的集群运行，这些brokers可以跨越多个数据中心或云区域。

这种集群架构支持对可扩展性、一致性、可用性、分区容错性和性能的需求。

在本文中，我们的学习目标是探索 Kafka 在 Kubernetes 上的可用性。

在此，特别说明，我们将独立设计一个 高可用的基于kubernetes的Kafka 集群：

1、优先考虑可用性而不是一致性，这是大家可能希望为实时指标收集等用例做出的权衡，在这种情况下，如果发生故障，写入新数据的可用性比丢失一些历史数据点更重要。(https://en.wikipedia.org/wiki/CAP_theorem)

2、优先选择简单而不是其他非功能性需求（例如安全性、性能、效率等），从而更专注于 Kafka 和 Kubernetes。

3、本文也假设维护服务的计划外中断比基础设施故障更有可能发生。

考虑到这些因素，让我们首先讨论一个典型的高可用性 Kafka 集群——运行于主机上而非运行于Kubernetes集群中。

Kafka partitions 与 replication-factor

在 Kafka 中，消息被分类为topics，在集群中，每个topic都有一个唯一的名称。

例如，如果您构建一个聊天应用程序，您可能会为每个聊天室设置一个topic（例如“dave-tom-chat”）。

但是当消息的数量超过broker限制的大小时会发生什么？

Topic被分解成多个分区，每个分区都可以存在于 Kafka 集群中的一个单独节点上。

换句话说，来自单个topic的所有消息可以存储在不同的broker中，但来自单个分区的所有消息只能在同一个节点上找到。

· 如果一个topic包含所有消息，当设备上没有空间时它是如何工作的？

· Kafka 使用分区将记录分发给多个broker。

· 每个topic可以有不同数量的分区。来自单个分区的所有记录始终一起存储在节点上。

这种设计选择支持topic的并行化、可扩展性和高消息吞吐量。

下面还有更多内容，让我们继续往下看。

Topic配置了一个replication factor，它决定了每个分区的副本数。

如果一个集群只有一个topic和一个分区，则replication factor为 3 意味着存在三个分区：每个分区一个副本。

分区的所有副本都存在于不同的broker上，因此您不能配置比集群中的节点多的分区副本。

在前面的示例中，replication factor为 3，您应该期望 Kafka 集群中至少有三个节点。

但是 Kafka 如何让这些副本保持同步呢？

分区被区分成leader和follower角色，其中分区leader处理所有写入和读取，follower纯粹用于故障转移。

follower可以与leader同步（包含所有分区leader的消息，除了缓冲区窗口中的消息）或不同步。

所有同步副本的集合称为 ISR（in-sync replicas）。

这些是 Kafka 和replications的基础；让我们看看故障时会发生什么。

了解brokers故障

假设 Kafka 集群有 3 个broker，replication factor为 1。

集群中只有一个topic和一个分区。

当 broker 不可用时，分区也不可用，集群无法为消费者或生产者提供服务。

让我们通过将replication factor设置为 3 来改变它。

在这种情况下，每个broker都有一个分区的副本。

当broker不可用时会发生什么？

如果分区有额外的同步副本，其中一个将成为临时分区leader。

集群可以照常运行，消费者或生产者没有停机时间。

一个所有分区同步的 Kafka 集群丢失了一个broker。

两个分区中的一个将被提升为leader，集群将继续照常运行。

当有分区副本但它们不同步时怎么办？

在这种情况下，有两种选择：

0. 选择等待分区leader重新上线——牺牲可用性。
1. 允许不同步的副本成为临时分区leader——牺牲一致性。
2. 
   

分区不同步的 Kafka 集群失去了一个broker。

集群可以将其中一个不同步的副本提升为leader。但是，您可能会丢失一些消息。

或者，您可以等待broker恢复，但是，这样影响您服务的可用性。

上面我们已经讨论了一些失败场景，下面让我们看看如何处理它们。

如何解决或者减轻常见故障

您可能注意到一个分区应该有一个额外的同步副本 (ISR) 可用以在分区leader丢失后幸存下来。

因此，一个简单的集群大小至少得有两个最小同步副本大小为 2 的broker。

然而，这还不够。

如果你只有两个副本，然后失去了一个broker，同步副本大小会减少到 1，生产者和消费者都无法工作（即最小同步副本为 2）。

因此，broker的数量应该大于最小同步副本大小（即至少 3 个）。

您可以设置一个只有两个broker且最小同步副本大小为 2 的 Kafka 集群。

但是，当broker丢失时，集群将变得不可用，因为单个副本处于同步状态。

您应该配置一个 Kafka 集群，该集群的broker数量大于同步副本的大小。

在这种情况下，如果一个 broker 丢失，Kafka 集群仍然可以继续运行。

那你又该如何规划broker的位置了？

考虑到大都使用云服务托管的 Kafka 集群，因此最好在故障域（例如区域、区域、节点等）之间分布broker节点。

因此，如果您希望设计一个可以容忍一次计划内和一次计划外故障的 Kafka 集群，您至少应该考虑以下要求：

· 最少 2 个同步副本。

· topic的replication factor为 3。

· 至少 3 个 Kafka broker，每个broker运行在不同的节点上。

· 节点分布在三个可用区。

在本文的剩余部分，您将在 Kubernetes 上构建和故障测试 Kafka 集群以验证这些假设。

在 Kubernetes 上部署 3 节点 Kafka 集群

让我们创建一个跨三个可用区的三节点集群：

$ k3d cluster create kube-cluster \
--agents 3 \
--k3s-node-label topology.kubernetes.io/zone=zone-a@agent:0 \
--k3s-node-label topology.kubernetes.io/zone=zone-b@agent:1 \
--k3s-node-label topology.kubernetes.io/zone=zone-c@agent:2
INFO[0000] Created network 'k3d-kube-cluster'
INFO[0000] Created image volume k3d-kube-cluster-imagesINFO[0000] Starting new tools node...INFO[0001] Creating node 'k3d-kube-cluster-server-0'
INFO[0003] Starting Node 'k3d-kube-cluster-tools'
INFO[0012] Creating node 'k3d-kube-cluster-agent-0'
INFO[0012] Creating node 'k3d-kube-cluster-agent-1'
INFO[0012] Creating node 'k3d-kube-cluster-agent-2'
INFO[0012] Creating LoadBalancer 'k3d-kube-cluster-serverlb'
INFO[0017] Starting new tools node...INFO[0017] Starting Node 'k3d-kube-cluster-tools'
INFO[0018] Starting cluster 'kube-cluster'
INFO[0018] Starting servers...INFO[0018] Starting Node 'k3d-kube-cluster-server-0'
INFO[0022] Starting agents...INFO[0022] Starting Node 'k3d-kube-cluster-agent-1'
INFO[0022] Starting Node 'k3d-kube-cluster-agent-0'
INFO[0022] Starting Node 'k3d-kube-cluster-agent-2'
INFO[0032] Starting helpers...INFO[0032] Starting Node 'k3d-kube-cluster-serverlb'
INFO[0041] Cluster 'kube-cluster' created successfully!

您可以通过以下方式验证集群是否已准备就绪：

$ kubectl get nodes
NAME                       STATUS   ROLES                 VERSION
k3d-kube-cluster-server-0   Ready   control-plane,master   v1.22.7+k3s1
k3d-kube-cluster-agent-1   Ready   <none>                 v1.22.7+k3s1
k3d-kube-cluster-agent-0   Ready   <none>                 v1.22.7+k3s1
k3d-kube-cluster-agent-2   Ready   <none>                 v1.22.7+k3s1

接下来，让我们将 Kafka 集群部署为 Kubernetes StatefulSet。

这是一个 YAML 清单，kafka.yaml定义了创建简单 Kafka 集群所需的资源：

apiVersion: v1
kind: Service
metadata:
name: kafka-svc
labels:
  app: kafka-app
spec:
clusterIP: None
ports:
  - name: '9092'
    port: 9092
    protocol: TCP
    targetPort: 9092
selector:
  app: kafka-app
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka
labels:
  app: kafka-app
spec:
serviceName: kafka-svc
replicas: 3
selector:
  matchLabels:
    app: kafka-app
template:
  metadata:
    labels:
      app: kafka-app
  spec:
    containers:
      - name: kafka-container
        image: doughgle/kafka-kraft
        ports:
          - containerPort: 9092
          - containerPort: 9093
        env:
          - name: REPLICAS
            value: '3'
          - name: SERVICE
            value: kafka-svc
          - name: NAMESPACE
            value: default
          - name: SHARE_DIR
            value: /mnt/kafka
          - name: CLUSTER_ID
            value: oh-sxaDRTcyAr6pFRbXyzA
          - name: DEFAULT_REPLICATION_FACTOR
            value: '3'
          - name: DEFAULT_MIN_INSYNC_REPLICAS
            value: '2'
        volumeMounts:
          - name: data
            mountPath: /mnt/kafka
volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      accessModes:
        - "ReadWriteOnce"
      resources:
        requests:
          storage: "1Gi"

您可以通过以下方式应用此 YAML 文件中的所有资源：

$ kubectl apply -f kafka.yaml
service/kafka-svc created
statefulset.apps/kafka created

检查创建的资源：

$ kubectl get -f kafka.yaml
NAME               TYPE       CLUSTER-IP   EXTERNAL-IP   PORT(S)
service/kafka-svc   ClusterIP   None         <none>       9092/TCP

NAME                     READY
statefulset.apps/kafka   3/3

有一个 StatefulSet，其中包含三个准备好的 Kafka broker pod 和一个服务。

还有三个独立的 PersistentVolumeClaims 用于存储 Kafka 数据，每个 broker 一个：

$ kubectl get pvc,pv
NAME                                 STATUS   VOLUME         CAPACITY   ACCESS MODES
persistentvolumeclaim/data-kafka-0   Bound   pvc-eec953ae   1Gi       RWO
persistentvolumeclaim/data-kafka-1   Bound   pvc-5544a431   1Gi       RWO
persistentvolumeclaim/data-kafka-2   Bound   pvc-11a64b48   1Gi       RWO

上面创建这些资源是什么？

让我们看一下kafka.yaml清单中配置的一些配置信息。

定义了两种资源：

1、一个StatefulSet

2、一个Headless service

KAFKA StatefulSet

StatefulSet 目的是为了创建 pod 副本的对象——就像deployment一样。

但与 Deployment 不同的是，StatefulSet 保障了 Pod 的顺序和唯一性。

StatefulSet是为了解决有状态服务的问题（对应Deployments和ReplicaSets是为无状态服务而设计），其应用场景包括：

· 稳定的持久化存储，即Pod重新调度后还是能访问到相同的持久化数据，基于PVC来实现

· 稳定的网络标志，即Pod重新调度后其PodName和HostName不变，基于Headless Service（即没有Cluster IP的Service）来实现

· 有序部署，有序扩展，即Pod是有顺序的，在部署或者扩展的时候要依据定义的顺序依次依次进行（即从0到N-1，在下一个Pod运行之前所有之前的Pod必须都是Running和Ready状态），基于init containers来实现

· 有序收缩，有序删除（即从N-1到0）

StatefulSet 中的每个 Pod 都从 StatefulSet 的名称和 Pod 的序号生成主机名。

命名方式是$(statefulset name)-$(ordinal)。

在该环境中，StatefulSets 的名称是kafka，因此您应该有三个带有kafka-0, kafka-1,kafka-2名称的 pod。

让我们验证一下：

$ kubectl get pods
NAME     READY   STATUS   RESTARTSkafka-0   1/1     Running   0
kafka-1   1/1     Running   0
kafka-2   1/1     Running   0

删除`kafka-0时会发生什么？

Kubernetes 会创建新的kafka-3吗？

让我们测试一下：

$ kubectl delete pod kafka-0
pod "kafka-0" deleted

列出正在运行的 pod：

$ kubectl get pods
NAME     READY   STATUS   RESTARTSkafka-1   1/1     Running   0
kafka-2   1/1     Running   0
kafka-0   1/1     Running   0

Kubernetes 重新创建了同名的 Pod！

让我们检查一下 StatefulSet YAML 定义的其余部分。

apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka labels:
  app: kafka-app
spec:
serviceName: kafka-svc
replicas: 3
selector:
  matchLabels:
    app: kafka-app template:
  metadata:
    labels:
      app: kafka-app   spec:
    containers:
      - name: kafka-container
        image: doughgle/kafka-kraft
        ports:
          - containerPort: 9092
# truncated output

StatefulSet 定义了三个副本，以便从 pod spec template创建三个 pod。

有容器从镜像启动时：

1、server.properties使用其唯一的broker ID、内部和外部listener以及仲裁投票者列表配置broker。

2、格式化日志目录。

3、启动 Kafka Java 进程。

容器镜像暴露了两个端口：

· 9092用于客户端通信。这对于生产者和消费者的连接是必要的。

· 9093用于内部、inter-broker间的通信。

在 YAML 的下一部分中，还包含一长串环境变量：

apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka
labels:
  app: kafka-app
spec:
serviceName: kafka-svc
replicas: 3
selector:
  matchLabels:
    app: kafka-app
template:
  metadata:
    labels:
      app: kafka-app
  spec:
    containers:
      - name: kafka-container
        image: doughgle/kafka-kraft
        ports:
          - containerPort: 9092
          - containerPort: 9093
        env:
          - name: REPLICAS
            value: '3'
          - name: SERVICE
            value: kafka-svc
          - name: NAMESPACE
            value: default
          - name: SHARE_DIR
            value: /mnt/kafka
          - name: CLUSTER_ID
            value: oh-sxaDRTcyAr6pFRbXyzA
          - name: DEFAULT_REPLICATION_FACTOR
            value: '3'
          - name: DEFAULT_MIN_INSYNC_REPLICAS
            value: '2'
        volumeMounts:
          - name: data
            mountPath: /mnt/kafka
volumeClaimTemplates:
# truncated output

这些在entry point脚本中用于broker设置的值server.properties（https://docs.confluent.io/platform/current/installation/configuration/broker-configs.html）：

· REPLICAS- 用作迭代器边界以将controller.quorum.voters（https://kafka.apache.org/documentation/#brokerconfigs_controller.quorum.voters）属性设置为broker列表。

· SERVICE和NAMESPACE- 用于为集群中的每个broker的 CoreDNS 解析以进行设置controller.quorum.voters（https://kafka.apache.org/documentation/#brokerconfigs_controller.quorum.voters），listeners（https://kafka.apache.org/documentation/#connectconfigs_listeners）和advertised.listeners（https://kafka.apache.org/documentation/#brokerconfigs_advertised.listeners）

· SHARE_DIR- 用于设置log.dirs（https://kafka.apache.org/documentation/#brokerconfigs_log.dirs）；存储 Kafka 数据的目录

· CLUSTER_ID是 Kafka 集群的唯一标识符

· DEFAULT_REPLICATION_FACTOR是集群范围的默认replication factor

· DEFAULT_MIN_INSYNC_REPLICAS是集群默认同步副本大小

在 YAML 的其余部分中，有 PersitentVolumeClaim 模板的定义和volumeMounts：

apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka
labels:
  app: kafka-app
spec:
serviceName: kafka-svc
replicas: 3
selector:
  matchLabels:
    app: kafka-app
template:
  metadata:
    labels:
      app: kafka-app
  spec:
    containers:
      - name: kafka-container
        image: doughgle/kafka-kraft
        ports:
          - containerPort: 9092
          - containerPort: 9093
        env:
           # truncated output
        volumeMounts:
          - name: data
            mountPath: /mnt/kafka
volumeClaimTemplates:
  - metadata:
      name: data
    spec:
      accessModes:
        - "ReadWriteOnce"
      resources:
        requests:
          storage: "1Gi"

对于每个 pod，StatefulSet 使用volumeClaimTemplates。

在这种情况下，它会创建一个 PersistentVolumeClaim，其中包含：

· ReadWriteOnce访问模式强制执行卷一次只能属于一个节点的约束

· 1Gi的存储

然后 PersistentVolumeClaim 通过 PersistentVolume 绑定到底层存储。

声明作为卷挂载在容器中，位置位于/mnt/kafka。

这是 Kafka broker将数据存储在按topic和分区组织的文件中的地方。

重要的是要注意 StatefulSet 保证给定的 Pod 将始终映射到相同的存储标识。

如果 Podkafka-0被删除，Kubernetes 会重新创建一个同名的 Pod，并挂载相同的 PersistentVolumeClaim 和 PersistentVolume。

请记住这一点，因为它将在以后变得有帮助。

将 StatefulSet 与 Headless Service相结合

在您的 Kafka 集群的 YAML 定义的开头，有一个 Service 定义：

apiVersion: v1
kind: Service
metadata:
name: kafka-svc
labels:
  app: kafka-app
spec:
clusterIP: None
ports:
  - name: '9092'
    port: 9092
    protocol: TCP
    targetPort: 9092
selector:
  app: kafka-app

上述内容包含的clusterIP: None通常称为Headless Service。

Kubernetes 有四种类型的服务：(https://kubernetes.io/docs/concepts/services-networking/service/)

1、ClusterIP

2、NodePort

3、LoadBalancer

4、External

那么，什么是Headerless Service？

Headless Service是没有 IP 地址的 ClusterIP 服务。

那么，你如何使用它呢？

Headless Service与 CoreDNS 结合使用会很有帮助。

当您向标准 ClusterIP 服务发出 DNS 查询时，您会收到一个 IP 地址：

$ dig standard-cluster-ip.default.svc.cluster.local

;; QUESTION SECTION:
;standard-cluster-ip.default.svc.cluster.local. IN   A

;; ANSWER SECTION:
standard-cluster-ip.default.svc.cluster.local. 30 IN A   10.100.0.1

但是，当您查询 Headless Service时，DNS 会使用 Pod 的所有单独 IP 地址进行回复（在这种情况下，该服务有两个 pod）：

$ dig headless.default.svc.cluster.local

;; QUESTION SECTION:
;headless.default.svc.cluster.local. IN A

;; ANSWER SECTION:headless.default.svc.cluster.local. 13 IN
A 10.0.0.1
headless.default.svc.cluster.local. 13 IN
A 10.0.0.2

所以，顾名思义，Headless Service就是没头的Service。有什么使用场景呢？

第一种：自主选择权，有时候client想自己来决定使用哪个Real Server，可以通过查询DNS来获取Real Server的信息。

第二种：Headless Services还有一个用处（PS：也就是我们需要的那个特性）。Headless Service的对应的每一个Endpoints，即每一个Pod，都会有对应的DNS域名；这样Pod之间就可以互相访问。

这如何与 StatefulSet 一起工作？

1、StatefulSet 将 pod 的名称设置为其主机名（例如kafka-0、kafka-1等）。

2、每个 Pod 都有一个可选subdomain字段，可用于指定其 DNS 子域。(https://kubernetes.io/docs/concepts/services-networking/dns-pod-service/#pod-s-hostname-and-subdomain-fields)

3、StatefulSet 在创建 Pod 时以 $(podname).$(governing service domain)的形式分配一个子域，其中 serviceName 字段定义了 StatefulSet 上的管理服务。

4、Pod 现在可以使用完全路径名称...svc.cluster.local.

例如，如果 Pod 的主机名设置为kafka-1，子域设置为kafka-svc，在 namespace 中default，将具有完全路径的域名 (FQDN) kafka-1.kafka-svc.default.svc.cluster.local。

现在我们已经进行了一些理论的学习，下面让我们通过发送消息来测试 Kafka 集群。

发布测试事件

在 Kafka 术语中，生产者可以将事件发布到topic。消费者可以订阅这些主题并使用这些事件。

让我们将一个简单的事件发布到一个topic并使用它。

在与容器交互之前，让我们通过描述Headless Service来查找broker的 IP 地址：

$ kubectl describe service kafka-svc
Name:             kafka-svc
Namespace:         default
Labels:           app=kafka-app
Selector:         app=kafka-app
Type:             ClusterIP
Port:             9092 9092/TCP
TargetPort:       9092/TCP
Endpoints:         10.42.0.10:9092,10.42.0.12:9092,10.42.0.13:9092

现在，让我们创建一个可以用作 Kafka 客户端的 pod：

$ kubectl run kafka-client --rm -ti --image bitnami/kafka:3.1.0 -- bash
I have no name!@kafka-producer:/$

在 Kafka 客户端容器中，有一组脚本可以更轻松进行以下的任务：

· 模拟生产者或消费者。

· 触发leader选举。

· 验证副本。

列出的内容如下：

$ ls /opt/bitnami/kafka/bin
kafka-acls.sh
kafka-broker-api-versions.sh
kafka-cluster.sh
kafka-configs.sh
kafka-console-consumer.sh
kafka-console-producer.sh
kafka-consumer-groups.sh
kafka-consumer-perf-test.sh
kafka-delegation-tokens.sh
kafka-delete-records.sh
# truncated output

使用“test” topic，让我们运行示例控制台生产者脚本kafka-console-producer：

$ kafka-console-producer.sh \
--topic test \
--request-required-acks all \
--bootstrap-server 10.42.0.10:9092,10.42.0.12:9092,10.42.0.13:9092

当>提示变为可见时，您可以生成“hello world”事件：

>hello world

注意脚本内容：

1、需要所有同步副本的确认才能提交一批消息。

2、Kafka broker的IP地址和端口号以逗号分隔。

事件存储在 Kafka 中，但消费者应该如何检索它？

消费“test” topic的事件

在同一个终端会话中，使用Ctrl+C终止脚本并运行消费者脚本：

$ kafka-console-consumer.sh \
--topic test \
--from-beginning \
--bootstrap-server 10.42.0.10:9092,10.42.0.12:9092,10.42.0.13:9092
hello world
^CProcessed a total of 1 messages

消费者继续轮询broker以获取有关该test topic的更多事件并在它们发生时对其进行处理。

您向该topic发布了一个“hello world”事件test，另一个进程使用了它。

下面我来考虑一下：

当工作节点上有维护活动时会发生什么？

它如何影响我们的 Kafka 集群？

让节点停机以进行维护：drain leader所在的节点

让我们模拟替换托管broker所在的 Kubernetes 节点。

首先，从 Kafka 客户端，让我们确定哪个broker是该testtopic的leader。

您可以使用kafka-topics.sh脚本描述topic：

$ kafka-topics.sh --describe \
--topic test \
--bootstrap-server 10.42.0.10:9092,10.42.0.12:9092,10.42.0.13:9092
Topic: test
TopicId: P0SP1tEKTduolPh4apeV8Q
PartitionCount: 1
ReplicationFactor: 3
Configs: min.insync.replicas=2,segment.bytes=1073741824

Topic: test
Partition: 0
Leader: 1
Replicas: 1,0,2
Isr: 1,0,2

Leader: 1表示该testtopic的leader是broker 1。

在这个 Kafka 设置中（按照上述约定），它的 pod 名称是kafka-1.

因此，既然您知道testtopic的leader在kafka-1pod 上，您应该找出该 pod 的部署位置：

$ kubectl get pod kafka-1 -o wide
NAME     READY   STATUS   RESTARTS   IP           NODE
kafka-1   1/1     Running   0         10.42.0.12   k3d-kube-cluster-agent-0

Broker 1 位于 Kubernetes 工作节点上k3d-kube-cluster-agent-0。

让我们排空它以驱逐 pod：

$ kubectl drain k3d-kube-cluster-agent-0 \
--delete-emptydir-data \
--force \
--ignore-daemonsets
node/k3d-kube-cluster-agent-0 cordoned
evicting pod default/kafka-1
pod/kafka-1 evicted
node/k3d-kube-cluster-agent-0 evicted

leaderkafka-1按预期被驱逐。

由于broker平均分布在 Kubernetes 工作节点上，因此对一个节点的维护只会降低broker总数的一小部分。

生产者和消费者还在工作吗？

Kafka 集群还能用吗？

生产者和消费者能否继续照常工作？

让我们重新运行 kafka 控制台生产者脚本：

$ kafka-console-producer.sh \
  --topic test \
  --bootstrap-server 10.42.0.10:9092,10.42.0.12:9092,10.42.0.13:9092

在>提示符下，您可以使用以下命令生成另一个“hello world”事件

WARN Bootstrap broker 10.42.0.10:9092 (id: -2 rack: null) disconnected (org.apache.kafka.clients.NetworkClient)
>hello again, world

请注意其中一个bootstrap servers无法解析的警告。

尽管如此，你还是设法产生了另一条信息。

但是消费者能收到吗？

使用Ctrl+C终止命令并执行以下命令：

$ kafka-console-consumer.sh \
  --topic test \
  --from-beginning \
  --bootstrap-server 10.42.0.10:9092,10.42.0.12:9092,10.42.0.13:9092
hello world
hello again, world

发生了什么？

两条消息都是从 Kafka 集群中检索到的——它成功了！

现在停止交互式会话并再次查看topic test信息：

$ kafka-topics.sh --describe \
--topic test \
--bootstrap-server 10.42.0.10:9092,10.42.0.12:9092,10.42.0.13:9092
Topic: test
TopicId: QqrcLtJSRoufzOZqNc9KcQPartitionCount: 1
ReplicationFactor: 3
Configs: min.insync.replicas=2,segment.bytes=1073741824

Topic: test
Partition: 0
Leader: 2
Replicas: 1,2,0
Isr: 2,0

有几个有趣的细节：

1、topic leader现在是2（原为 1）。

2、同步副本列表Isr包含2,0- broker0 和borker 2。

3、但是，broker 1 不同步。

这是正确的，因为broker 1不再可用。

Kafka pod 处于待处理状态

所以一个节点停机维护，如果你列出所有正在运行的 Pod，你会注意到它kafka-0是 Pending状态。

$ kubectl get pod -l app=kafka-app
NAME     READY   STATUS   RESTARTS
kafka-0   1/1     Running   0
kafka-2   1/1     Running   0
kafka-1   0/1     Pending   0

但是 Kubernetes 不应该将 Pod 重新调度到另一个工作节点吗？

让我们通过描述 pod 进行查看详情：

$ kubectl describe pod kafka-1
# truncated
Events:
Type     Reason            From               Message
----     ------            ----               -------
Warning  FailedScheduling  default-scheduler  0/3 nodes are available:
                                              1 node(s) were unschedulable,
                                              3 node(s) had volume node affinity conflict.

kafka-1没有可用的节点。

虽然只是k3d-kube-cluster-agent-0为了维护而离线，但其他节点不满足持久卷的节点亲和性约束。

让我们验证一下。

首先，让我们找到绑定到 (defunct) 的 PersistentVolume kafka-1：

$ kubectl get persistentvolumes,persistentvolumeclaims
NAME                            CAPACITY   ACCESS MODES   RECLAIM POLICY   STATUS     CLAIM
persistentvolume/pvc-018e8d78   1Gi        RWO            Delete           Bound      default/data-kafka-1
persistentvolume/pvc-455a7f5b   1Gi        RWO            Delete           Bound      default/data-kafka-2
persistentvolume/pvc-abd6b6cf   1Gi        RWO            Delete           Bound      default/data-kafka-0

NAME                                 STATUS   VOLUME         CAPACITY   ACCESS MODES
persistentvolumeclaim/data-kafka-1   Bound    pvc-018e8d78   1Gi        RWO
persistentvolumeclaim/data-kafka-2   Bound    pvc-455a7f5b   1Gi        RWO
persistentvolumeclaim/data-kafka-0   Bound    pvc-abd6b6cf   1Gi        RWO

您可以使用以下命令检查 PersitentVolume：

kubectl get persistentvolume pvc-018e8d78
apiVersion: v1
kind: PersistentVolume
metadata:
  name: pvc-018e8d78
spec:
  accessModes:
  - ReadWriteOnce
  capacity:
    storage: 1Gi
  # truncated
  hostPath:
    path: /var/lib/rancher/k3s/storage/pvc-018e8d78_default_data-kafka-0
    type: DirectoryOrCreate
  nodeAffinity:
    required:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/hostname
          operator: In
          values:
          - k3d-kube-cluster-agent-0
  persistentVolumeReclaimPolicy: Delete
  storageClassName: local-path
  volumeMode: Filesystem只有需要`k3d-kube-cluster-agent-0`的音量`kafka-1`。

并且 PersistentVolume 不能移动到其他地方，因此任何需要访问该卷的 pod 都应该从k3d-kube-cluster-agent-0.

由于节点不可用，调度器无法分配 Pod，Pod 一直处于 Pending 状态。

请注意，此卷调度约束是由本地平台(https://github.com/rancher/local-path-provisioner)强加的，并非对所有供应商都通用。

换句话说，您可能会发现另一个provisioner 可以将PersistentVolume 附加到不同的节点，并且可以将Pod 重新调度到与另一个broker相同的节点上。

但这并不是很好——失去单个节点可能会损害 Kafka 集群的可用性。

让我们通过对 Pod 使用topology constraint引入一个约束来解决这个问题。(https://kubernetes.io/docs/concepts/workloads/pods/pod-topology-spread-constraints/)

Pod Topology Constraints帮助您跨故障域分布 Pod

在所有公共云中，一个区域将可能一起发生故障的资源分组，例如，由于意外断电的情况。

但是，不同区域中的资源不太可能同时出现故障。

这有助于确保恢复能力，因为一个区域的断电不会影响另一个区域。

尽管区域的确切定义是由基础设施决定的，但你可以想象两到三个机房，每个机房都有独立的空调、电源、网络交换机、机架等。

区域是故障域的一个场景。

另一个场景可能是一个地区。英国南部和美国东部地区不太可能同时失败。

在 Kubernetes 中，您可以使用此信息来设置 Pod 应放置在何处的约束。

例如，您可以将 Kafka broker限制在不同的区域中。

下面是一个如何做到这一点的例子：

apiVersion: apps/v1
kind: StatefulSet
metadata:
name: kafka
labels:
  app: kafka-app
spec:
serviceName: kafka-svc
replicas: 3
selector:
  matchLabels:
    app: kafka-app
template:
  metadata:
    labels:
      app: kafka-app
  spec:
    topologySpreadConstraints:
      - maxSkew: 1
        topologyKey: kubernetes.io/zone
        whenUnsatisfiable: DoNotSchedule
        labelSelector:
          matchLabels:
            app: kafka-app
    containers:
      - name: kafka-container
# truncated output

topologySpreadConstraints内容如下：

• maxSkew 描述 Pod 分布不均的程度。这是给定拓扑类型中任意两个拓扑域中 匹配的 pod 之间的最大允许差值。它必须大于零。取决于 whenUnsatisfiable 的 取值，其语义会有不同。
• 当 whenUnsatisfiable 等于 "DoNotSchedule" 时，maxSkew 是目标拓扑域 中匹配的 Pod 数与全局最小值之间可存在的差异。
• 当 whenUnsatisfiable 等于 "ScheduleAnyway" 时，调度器会更为偏向能够降低 偏差值的拓扑域。
• topologyKey 是节点标签的键。如果两个节点使用此键标记并且具有相同的标签值， 则调度器会将这两个节点视为处于同一拓扑域中。调度器试图在每个拓扑域中放置数量 均衡的 Pod。
• whenUnsatisfiable 指示如果 Pod 不满足分布约束时如何处理：
• DoNotSchedule（默认）告诉调度器不要调度。
• ScheduleAnyway 告诉调度器仍然继续调度，只是根据如何能将偏差最小化来对 节点进行排序。
• labelSelector 用于查找匹配的 pod。匹配此标签的 Pod 将被统计，以确定相应 拓扑域中 Pod 的数量。
• 
  

当 Pod 定义了不止一个 topologySpreadConstraint，这些约束之间是逻辑与的关系。kube-scheduler 会为新的 Pod 寻找一个能够满足所有约束的节点。

你可以执行 kubectl explain Pod.spec.topologySpreadConstraints 命令以 了解关于 topologySpreadConstraints 的更多信息。

在拓扑到位后，Pod 将始终分布在所有可用区域中——无论 PersistentVolume 中的节点亲和性如何。

节点亲和性Node Affinity

Affinity 翻译成中文是“亲和性”，它对应的是 Anti-Affinity，我们翻译成“互斥”。这两个词比较形象，可以把 pod 选择 node 的过程类比成磁铁的吸引和互斥，不同的是除了简单的正负极之外，pod 和 node 的吸引和互斥是可以灵活配置的。

kubernetes 1.2 版本开始引入这个概念，目前（1.6版本）处于 beta 阶段，相信后面会变成核心的功能。这种方法比 nodeSelector 复杂，但是也更灵活，提供了更精细的调度控制。它的优点包括：

· 匹配有更多的逻辑组合，不只是字符的完全相等

· 调度分成软策略（soft）和硬策略（hard），在软策略的情况下，如果没有满足调度条件的节点，pod 会忽略这条规则，继续完成调度过程

目前有两种主要的 node affinity： requiredDuringSchedulingIgnoredDuringExecution 和 preferredDuringSchedulingIgnoredDuringExecution。前者表示 pod 必须部署到满足条件的节点上，如果没有满足条件的节点，就不断重试；后者表示优先部署在满足条件的节点上，如果没有满足条件的节点，就忽略这些条件，按照正常逻辑部署。

IgnoredDuringExecution 正如名字所说，pod 部署之后运行的时候，如果节点标签发生了变化，不再满足 pod 指定的条件，pod 也会继续运行。与之对应的是 requiredDuringSchedulingRequiredDuringExecution，如果运行的 pod 所在节点不再满足条件，kubernetes 会把 pod 从节点中删除，重新选择符合要求的节点。

我们先从常见的几个 Pod 定义 NodeAffinity 亲和实例开始，熟悉一下 NodeAffinity 配置定义

实例一（matchExpressions） 实现目标：多区域部署应用

apiVersion:v1
kind: Pod
metadata:
name: with-node-affinity
spec:
affinity:
  nodeAffinity:                                #pod实例部署在az1 或 az2
    requiredDuringSchedulingIgnoredDuringExecution:
      nodeSelectorTerms:
      - matchExpressions:
        - key: kubernetes.io/e2e-az-name       #node标签可自定义匹配
          operator: In
          values:
          - e2e-az1
          - e2e-az2

实例二（matchFields） 实现目标：排除指定节点部署应用

apiVersion: v1
kind: Pod
metadata:
name: nginx
spec:
containers:
- name: nginx
  image: nginx
affinity:
  nodeAffinity:
    requiredDuringSchedulingIgnoredDuringExecution:
      nodeSelectorTerms:
      - matchFields:
        - key: metadata.name                        #Node name
          operator: NotIn
          values:
          - work-node-abc

您应该使用 Pod 拓扑约束还是节点亲和性？

将卷（和POD）分配给Kubernetes节点的想法合理吗？Kubernetes背后的想法不是说POD可以在集群中的任何地方重新调度吗？虽然这可能适用于无状态应用程序，但对于Kafka这样的有状态应用程序，情况更为复杂。

有状态的应用程序具有独特的要求，例如：

1、您希望控制哪些应用程序与有状态应用程序共享计算资源，以避免共享计算资源pod的频繁变更情况。

2、理想情况下，数据不会从一个节点移动到另一个节点，因为跨节点复制 TB 数据速度很慢且容易出错。

3、您希望配置经过I/O优化的Kubernetes节点。

因此，将节点分配给 StatefulSet 并确保它们具有正确实例类型的专用节点池通常是一个比较好的建议。

根据您的设置，您可能希望将 NodeAffinity 分配给您的卷(https://kubernetes.io/docs/concepts/storage/persistent-volumes/#node-affinity) 和 Pod。(https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/)

全力恢复

假设您已完成对节点的维护，并且准备好uncordon节点。

$ kubectl uncordon k3d-kube-cluster-agent-0
node/k3d-kube-cluster-agent-0 uncordoned

现在 Kubernetes 可以将kafka-1Pod 调度到k3d-kube-cluster-agent-0与其 Persistent Volume 关联。

片刻之后，您应该注意到 pod 正在运行：

$ kubectl get pod kafka-1 -o wide
NAME     READY   STATUS   IP           NODE
kafka-1   1/1     Running   10.42.0.14   k3d-kube-cluster-agent-0

broker是否恢复成为集群的一部分？

由于重新创建了 pod 并分配了不同的 IP 地址，您应该使用以下命令检索新的endpoint列表：

$ kubectl describe service kafka-svc
Name:             kafka-svc
Namespace:         default
Labels:           app=kafka-app
Selector:         app=kafka-app
Type:             ClusterIP
Port:             9092 9092/TCP
TargetPort:       9092/TCP
Endpoints:         10.42.0.12:9092,10.42.0.14:9092,10.42.0.13:9092

更新bootstrap broker端点列表，并使用以下命令查询测试topic：

$ kafka-topics.sh --describe \
 --topic test \
 --bootstrap-server 10.42.0.12:9092,10.42.0.14:9092,10.42.0.13:9092
Topic: test
TopicId: QqrcLtJSRoufzOZqNc9KcQ
PartitionCount: 1
ReplicationFactor: 3
Configs: min.insync.replicas=2,segment.bytes=1073741824

Topic: test
Partition: 0
Leader: 2
Replicas: 1,2,0
Isr: 2,0,1

请注意，同步副本的列表是2,0,1。

因此kafka-1pod（broker 1）可以重新加入 Kafka 集群并关注附加消息hello again, world！

但是，在此示例中，您仅删除了一个节点。

如果多个节点同时下线会怎样？

多个节点停机以进行维护

想象一下，当其中一个节点已经离线，此时集群管理员手工或者自动下线了另外一个节点时候。会发生什么？

现在，让我们测试一下：

$ kubectl drain k3d-kube-cluster-agent-0 \
  --delete-emptydir-data \
  --force \
  --ignore-daemonsets
node/k3d-kube-cluster-agent-0 cordoned
evicting pod default/kafka-1
pod/kafka-1 evicted
node/k3d-kube-cluster-agent-0 evicted

$ kubectl drain k3d-kube-cluster-agent-1 \
  --delete-emptydir-data \
  --force \
  --ignore-daemonsets
node/k3d-kube-cluster-agent-1 cordoned
evicting pod default/kafka-2
pod/kafka-2 evicted
node/k3d-kube-cluster-agent-1 evicted

节点k3d-kube-cluster-agent-0和k3d-kube-cluster-agent-1被drained，podkafka-1与kafka-2被驱逐。

现在，如果您查询 pod：

$ kubectl get pod -l app=kafka-app
NAME     READY   STATUS   RESTARTS
kafka-1   1/1     Pending   1
kafka-0   0/1     Running   0
kafka-2   0/1     Pending   0

只有一个broker运行，生产者和消费者能否继续照常工作？

让我们运行 kafka 控制台生产者脚本并产生另一条消息：

$ kafka-console-producer.sh \
--topic test \
--bootstrap-server 10.42.0.12:9092,10.42.0.14:9092,10.42.0.13:9092

在>提示符下，我们可以生成另外一条hello?, world?!消息。

$ kafka-console-producer.sh \
--topic test \
--request-required-acks all \
--bootstrap-server 10.42.0.12:9092,10.42.0.14:9092,10.42.0.13:9092
>hello? world?!
WARN [Producer clientId=console-producer] NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender)
WARN [Producer clientId=console-producer] NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender)
WARN [Producer clientId=console-producer] NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender)
ERROR Messages are rejected since there are fewer in-sync replicas than required.

生产者被阻塞，因为同步副本少于所需的数量。

消费者呢？

让我们测试一下：

$ kafka-console-consumer.sh \
  --topic test \
  --from-beginning \
  --bootstrap-server 10.42.0.12:9092,10.42.0.14:9092,10.42.0.13:9092
WARN [Producer clientId=console-producer] NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender)
WARN [Producer clientId=console-producer] NOT_ENOUGH_REPLICAS (org.apache.kafka.clients.producer.internals.Sender)
ERROR Messages are rejected since there are fewer in-sync replicas than required.

它似乎也被阻止消费。

Kafka正式不可用。

让我们修复它，这样这种故障就不会再发生了。

Pod Disruption Budget

您可以使用 Pod Disruption Budget (PDB) 来限制因维护而造成的中断。

PodDisruptionBudgets 定义了该应用程序运行所需的最小副本数。

在 Kubernetes 中，为了保证业务不中断或业务SLA不降级，需要将应用进行集群化部署。通过PodDisruptionBudget 控制器可以设置应用POD集群处于运行状态最低个数，也可以设置应用POD集群处于运行状态的最低百分比，这样可以保证在主动销毁应用POD的时候，不会一次性销毁太多的应用POD，从而保证业务不中断或业务SLA不降级。

PodDisruptionBudget 如下所示：

apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
name: kafka-pdb
spec:
minAvailable: 2
selector:
  matchLabels:
    app: kafka-app

有三个字段：

· 与 Pod 匹配的标签选择器。

· spec.minAvailable：表示发生自愿中断的过程中，要保证至少可用的Pods数或者比例

· spec.maxUnavailable：表示发生自愿中断的过程中，要保证最大不可用的Pods数或者比例

上面配置只能用来对应 Deployment，RS，RC，StatefulSet的Pods，推荐优先使用 .spec.maxUnavailable。

注意：

· 同一个 PDB Object 中不能同时定义 .spec.minAvailable 和 .spec.maxUnavailable。

· 前面提到，应用滚动更新时Pod的delete和unavailable虽然也属于自愿中断，但是实际上滚动更新有自己的策略控制（marSurge 和 maxUnavailable），因此PDB不会干预这个过程。

· PDB 只能保证自愿中断时的副本数，比如 evict pod过程中刚好满足 .spec.minAvailable 或 .spec.maxUnavailable，这时某个本来正常的Pod突然因为Node Down(非自愿中断)挂了，那么这个时候实际Pods数就比PDB中要求的少了，因此PDB不是万能的！

使用上，如果设置 .spec.minAvailable 为 100% 或者 .spec.maxUnavailable 为 0%，意味着会完全阻止 evict pods 的过程（ Deployment和StatefulSet的滚动更新除外 ）。

由于 Kafka 集群应保持实例数量高于仲裁，您将设置minAvailable为 2。

让我们创建对象：

$ kubectl apply -f kafka-pdb.yaml
poddisruptionbudget.policy/kafka-pdb created

要测试 PodDisruptionBudget，您应该至少运行两个 Pod，并尝试将数量减少至少一个。

由于您有一个 Pod 正在运行，因此让我们使用以下命令uncordon其中一个节点：

$ kubectl uncordon k3d-kube-cluster-agent-0
node/k3d-kube-cluster-agent-0 uncordoned

可以发现应该有两个 Pod ，一个正在运行，另外一个处于 Pending：

$ kubectl get pod -l app=kafka-app
NAME      READY   STATUS    RESTARTS
kafka-1   1/1     Running   1
kafka-0   1/1     Running   0
kafka-2   0/1     Pending   0

如果驱逐事件超出 pod PodDisruptionBudget，则将防止中断。

让我们通过再次排空节点来测试：

$ kubectl drain k3d-kube-cluster-agent-0 \
  --delete-emptydir-data \
  --force \
  --ignore-daemonsets
node/k3d-kube-cluster-agent-0 cordoned
evicting pod default/kafka-0
error when evicting pods/"kafka-0" -n "default" (will retry after 5s):
Cannot evict pod as it would violate the pod's disruption budget.

尽管出现错误，但请注意该节点仍处于cordoned状态，以防止 Kubernetes 在其上调度新的 Pod：

$ kubectl get nodes
NAME                       STATUS                     ROLES                 VERSION
k3d-kube-cluster-server-0   Ready                       control-plane,master   v1.22.7+k3s1
k3d-kube-cluster-agent-1   Ready,SchedulingDisabled   <none>                 v1.22.7+k3s1
k3d-kube-cluster-agent-0   Ready,SchedulingDisabled   <none>                 v1.22.7+k3s1
k3d-kube-cluster-agent-2   Ready                       <none>                 v1.22.7+k3s1

但 Kafka pod 仍在运行：

$ kubectl get pods -o wide
NAME     READY   STATUS   IP           NODE
kafka-1   1/1     Running   10.42.0.15   k3d-kube-cluster-agent-0
kafka-0   1/1     Running   10.42.0.13   k3d-kube-cluster-agent-2
kafka-2   1/1     Pending   <none>       <none>

此外，不受 PodDisruptionBudget 约束的 pod 仍将被驱逐和重新调度。

节点k3d-kube-cluster-agent-1仍然不可用；如果它恢复不了怎么办？

灾难故障：节点永久下线！

如果使用两个同步副本复制所有分区，则如果永久删除Kubernetes节点，则不应丢失任何数据。但是，由于持久卷上的节点亲和性限制，broker pod永远不会被重新调度。让我们探索一下会出现什么问题。

您可以使用以下方法完全删除节点：

$ kubectl delete node k3d-kube-cluster-agent-1
node "k3d-kube-cluster-agent-1" deleted

在这种情况下，kafka-2pending是因为k3d-kube-cluster-agent-1已经消失了，并且随之而来的是 kafka-2 的本地数据。

$ kubectl get pods kafka-1 -o wide
NAME     READY   STATUS   RESTARTS
kafka-2   0/1     Pending   0

它不能被重新调度到另一个节点上，因为没有其他节点可以满足卷上的 nodeAffinity 约束。

生产者和消费者还在工作吗？

集群可以忍受这个吗？

在运行两个broker的情况下，您可能希望 Kafka 依然可供生产者和消费者使用。

让我们做一个快速的全面检查。

由于大多数broker IP 地址已被轮换，让我们通过以下方式验证它们：

$ kubectl describe service kafka-svc
Name:             kafka-svc
Namespace:         default
Labels:           app=kafka-app
Selector:         app=kafka-app
Type:             ClusterIP
Port:             9092 9092/TCP
TargetPort:       9092/TCP
Endpoints:         10.42.0.15:9092,10.42.0.13:9092

生产者发送一条信息，“Hello World. Do you copy?”：

$ kafka-console-producer.sh \
--topic test \
--request-required-acks all \
--bootstrap-server 10.42.0.15:9092,10.42.0.13:9092
>Hello World. Do you copy?

该消息似乎已提交。

消费新消息和历史消息怎么样？

$ kafka-console-consumer.sh \
--topic test \
--from-beginning \
--bootstrap-server 10.42.0.15:9092,10.42.0.13:9092
hello world
hello again, world
Hello World. Do you copy?

消费者能够消费所有消息！

让我们也检查一下“test” topic：

$ kafka-topics.sh --describe \
--topic test \
--bootstrap-server 10.42.0.15:9092,10.42.0.13:9092
Topic: test
TopicId: QqrcLtJSRoufzOZqNc9KcQ
PartitionCount: 1
ReplicationFactor: 3
Configs: min.insync.replicas=2,segment.bytes=1073741824

Topic: test
Partition: 0
Leader: 1
Replicas: 1,2,0
Isr: 1,2

所以，生产者和消费者仍然可用，但我们可以在 Kafka 集群中只使用两个broker节点吗？不，不是，我们不推荐这样的操作。

当前状态禁止了所有的维护操作。

因为我们已经对节点进行了像drain这样的操作，例如：

$ kubectl drain k3d-kube-cluster-agent-2 --ignore-daemonsets
node/k3d-kube-cluster-agent-2 cordoned
evicting pod default/kafka-0
error when evicting pods/"kafka-0" -n "default" (will retry after 5s):
Cannot evict pod as it would violate the pod's disruption budget.

那么，我们如何从这种情况中恢复过来呢？

Kafka-2 下线，它的继任者新 kafka-2 上线

您可以在节点故障的同一区域 (zone-a) 中添加新的 Kubernetes 工作节点k3d-kube-cluster-agent-1。

在本教程中，使用 k3d 添加新节点如下所示：

$ k3d node create kube-cluster-new-agent \
--cluster kube-cluster \
--k3s-node-label topology.kubernetes.io/zone=zone-b
INFO[0000] Adding 1 node(s) to the runtime local cluster 'kube-cluster'...
INFO[0000] Starting Node 'k3d-kube-cluster-new-agent-4'
INFO[0008] Successfully created 1 node(s)!

您可以使用kubectl get nodes它来查看它是否加入了集群：

$ kubectl get nodes
NAME                           STATUS     VERSION
k3d-kube-cluster-new-agent-4   Ready     v1.21.5+k3s2
# truncated output

状态如上——加入并准备就绪。

您可以通过删除 PVC 来清理旧broker数据：

$ kubectl delete pvc data-kafka-2
persistentvolumeclaim "data-kafka-0" deleted

当我们删除kafka-2pod 时，kubernetes 可以将其重新调度到新节点。

$ kubectl delete po kafka-2
pod "kafka-2" deleted

您可以实时观察新的 Kafka broker pod bootstrap：

$ kubectl get pods --watch
NAME             READY   STATUS             RESTARTS   AGE
kafka-0           1/1     Running             1         4d23h
kafka-1           1/1     Running             8         14d
kafka-2           0/1     ContainerCreating   0         14s
kafka-2           1/1     Running             0         51s

如果您检查状态，您会注意到所有 Kafka broker都在运行，并创建了一个新的 PersistentVolumeClaim 和 PersistentVolume：

$ kubectl get pods,pvc,pv
NNAME                 READY   STATUS
pod/kafka-2           1/1     Running
pod/kafka-1           1/1     Running
pod/kafka-0           1/1     Running

NAME                                 STATUS   VOLUME         CAPACITY   ACCESS MODES
persistentvolumeclaim/data-kafka-1   Bound   pvc-018e8d78   1Gi       RWO
persistentvolumeclaim/data-kafka-2   Bound   pvc-455a7f5b   1Gi       RWO
persistentvolumeclaim/data-kafka-0   Bound   pvc-abd6b6cf   1Gi       RWO

NAME                           CAPACITY   ACCESS MODES   RECLAIM POLICY   STATUS     CLAIM
persistentvolume/pvc-018e8d78   1Gi       RWO           Delete           Bound     default/data-kafka-1
persistentvolume/pvc-455a7f5b   1Gi       RWO           Delete           Bound     default/data-kafka-2
persistentvolume/pvc-fe291ef2   1Gi       RWO           Delete           Released   default/data-kafka-0
persistentvolume/pvc-abd6b6cf   1Gi       RWO           Delete           Bound     default/data-kafka-0

新的broker数据是否同步？

随着test topic分区被复制了 3 次，您应该期望kafka-0最终与其他broker同步。

使用以下命令检索新端点：

$ kubectl describe service kafka-svc
Name:             kafka-svc
Namespace:         default
Labels:           app=kafka-app
Selector:         app=kafka-app
Type:             ClusterIP
Port:             9092 9092/TCP
TargetPort:       9092/TCP
Endpoints:         10.42.0.15:9092,10.42.0.13:9092,10.42.1.16:9092

让我们查询test topic并检查状态：

$ kafka-topics.sh --describe \
--topic test \
--bootstrap-server 10.42.0.15:9092,10.42.0.13:9092,10.42.1.16:9092
Topic: test
TopicId: QqrcLtJSRoufzOZqNc9KcQ
PartitionCount: 1
ReplicationFactor: 3
Configs: min.insync.replicas=2,segment.bytes=1073741824

Topic: test
Partition: 0
Leader: 2
Replicas: 1,2,0
Isr: 2,0,1

是的，test topic有“2,0,1”作为同步副本，这意味着所有broker都与test topic同步。让我们使用来自新broker的消息来验证这一点。

首先，让我们获取kafka-0最新broker的 Pod IP 地址。

$ kubectl get pod kafka-0 -o jsonpath='{.status.podIP}'
10.42.0.13

其次，让我们从 Kafka 客户端 pod 运行控制台消费者，仅指定kafka-0's pod IP 地址：

$ kafka-console-consumer.sh \
--topic test \
--from-beginning \
--bootstrap-server 10.42.0.13:9092
hello worldhello again, worldHello World. Do you copy?

结果显而易见，数据是同步的！

概括

在本文中，我们一起设计并测试了 Kafka 集群的高可用性，以便生产者和消费者在发生故障时可以继续工作。

· 我们必须得承认100% 的可用性是不可能的，并且因维护和意外导致的中断要比基础设施的故障更有可能发生。

· Kubernetes 丢失了一个节点及其持久卷。

· Kafka 最终恢复了，没有数据丢失，也没有阻塞生产者和消费者。

· Kafka 可以利用本地存储来存储其数据，因为它负责topic分区和副本。

· 我们还了解了如何利用 Kubernetes 功能使您的 Kafka 集群具有高可用性：

· 您使用 pod 拓扑约束将节点分布在节点和可用区之间。

· 您将了解local path provisioner如何使用 nodeAffinity 创建持久卷。

· 您定义了 pod PDB，以防止可能导致计划外停机的管理事件。

请注意，为了简单起见，本文使用了 Kraft 模式(https://developer.confluent.io/learn/kraft/)（又名 Zookeeperless）的 Kafka，以便我们可以专注于 Kubernetes 中单个有状态服务的可用性。

然而，KRaft 还没有准备好在生产使用。

特别是，分区重新分配、不干净的leader选举、动态更改broker端点以及所有类型的升级在 Kraft 模式下都是不支持的。

原文：https://learnk8s.io/kafka-ha-kubernetes

参考：

0. Kubernetes Documentation | Kubernetes(https://kubernetes.io/docs/home/)
1. Apache Kafka(https://kafka.apache.org/documentation/)