概述

Apache ZooKeeper 是一个集中式服务，用于维护配置信息、命名、提供分布式同步和提供组服务，ZooKeeper 致力于开发和维护一个开源服务器，以实现高度可靠的分布式协调，其实也可以认为就是一个分布式数据库，只是结构比较特殊，是树状结构。官网文档：https://zookeeper.apache.org/doc/r3.8.0/关于Zookeeper的介绍，

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区的（partition）、多副本的（replica），基于zookeeper协调的分布式消息系统。官方文档：https://kafka.apache.org/documentation

安装helm

文件下载

链接：https://pan.baidu.com/s/1K1IzPjygF4-lepAA-7O2uA 
提取码：y30k 
--来自百度网盘超级会员V4的分享

安装helm

没有网络的情况下使用 helm pull 从有网络的机器上执行并copy内网。

helm pull bitnami/kafka

helm pull bitnami/zookeeper --untar

–untar 参数会直接加压出来

tar -zxvf helm-v3.0.0-linux-amd64.tar.gz
mv linux-amd64/helm /usr/local/bin/helm

参考: https://helm.sh/zh/docs/intro/install/

helm install zk

架构图

如果没有外网，文件下载中有zk和kafka的安装包使用. 一定要使用安装包中的镜像版本，否则会有问题.

]#helm install zookeeper -n kafka --set replicaCount=3  --set auth.enabled=false --set allowAnonymousLogin=true /opt/helm/zookeeper
NAME: zookeeper
LAST DEPLOYED: Fri Aug 12 14:06:02 2022
NAMESPACE: kafka
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
CHART NAME: zookeeper
CHART VERSION: 10.0.7
APP VERSION: 3.8.0
** Please be patient while the chart is being deployed **
ZooKeeper can be accessed via port 2181 on the following DNS name from within your cluster:
    zookeeper.kafka.svc.cluster.local
To connect to your ZooKeeper server run the following commands:
    export POD_NAME=$(kubectl get pods --namespace kafka -l "app.kubernetes.io/name=zookeeper,app.kubernetes.io/instance=zookeeper,app.kubernetes.io/component=zookeeper" -o jsonpath="{.items[0].metadata.name}")
    kubectl exec -it $POD_NAME -- zkCli.sh
To connect to your ZooKeeper server from outside the cluster execute the following commands:
    kubectl port-forward --namespace kafka svc/zookeeper 2181:2181 &
    zkCli.sh 127.0.0.1:2181

确认pod状态

]# kubectl get po -n kafka
NAME                             READY   STATUS    RESTARTS   AGE
kafka-manager-57569fcc78-mnhnn   1/1     Running   0          45h
zookeeper-0                      1/1     Running   0          10m
zookeeper-1                      1/1     Running   0          10m
zookeeper-2                      1/1     Running   0          10m

可以使用helm安装提示的命令测试一下

kafka的zk存储结构

zookeeper 是 kafka 不可分割的一部分。接下来就来讲讲zookeeper在kafka中作用。

1）记录和维护broker状态

zookeeper 记录了所有 broker 的存活状态，broker 会向 zookeeper 发送心跳请求来上报自己的状态。

zookeeper 维护了一个正在运行并且属于集群的 broker 列表。

2）控制器（leader ）选举

kafka 集群中有多个 broker，其中有一个会被选举为控制器。控制器负责管理整个集群所有分区和副本的状态，例如某个分区的 leader 故障了，控制器会选举新的 leader。从多个 broker 中选出控制器，这个工作就是 zookeeper 负责的。

3）限额权限

kafka 允许一些 client 有不同的生产和消费的限额，这些限额配置信息是保存在 zookeeper 里面的。

4）记录 ISR（已同步的副本）

Kafka会在Zookeeper上针对每个Topic维护一个称为ISR（in-sync replica，已同步的副本）的集合，该集合中是一些分区的副本。只有当这些副本都跟Leader中的副本同步了之后，kafka才会认为消息已提交，并反馈给消息的生产者。如果这个集合有增减，kafka会更新zookeeper上的记录。zookeeper 发现其中有成员不正常，马上移除。

5）node 和 topic 注册

zookeeper 保存了所有 node 和 topic 的注册信息，可以方便的找到每个 broker 持有哪些 topic。

node 和 topic 在 zookeeper 中是以临时节点的形式存在的，只要与 zookeeper 的 session 一关闭，他们的信息就没有了。

6）topic 配置

zookeeper 保存了 topic 相关配置，例如 topic 列表、每个 topic 的 partition 数量、副本的位置等等。

helm install kafka

[root@master2 /opt/helm/zookeeper]#helm install kafka -n kafka --set zookeeper.enabled=false --set replicaCount=3  --set externalZookeeper.servers=zookeeper /opt/helm/kafka/
NAME: kafka
LAST DEPLOYED: Fri Aug 12 14:18:47 2022
NAMESPACE: kafka
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
CHART NAME: kafka
CHART VERSION: 18.0.8
APP VERSION: 3.2.1
** Please be patient while the chart is being deployed **
Kafka can be accessed by consumers via port 9092 on the following DNS name from within your cluster:
    kafka.kafka.svc.cluster.local
Each Kafka broker can be accessed by producers via port 9092 on the following DNS name(s) from within your cluster:
    kafka-0.kafka-headless.kafka.svc.cluster.local:9092
    kafka-1.kafka-headless.kafka.svc.cluster.local:9092
    kafka-2.kafka-headless.kafka.svc.cluster.local:9092
To create a pod that you can use as a Kafka client run the following commands:
    kubectl run kafka-client --restart='Never' --image 10.50.10.185/kafka/docker.io/bitnami/kafka:3.2.1-debian-11-r4 --namespace kafka --command -- sleep infinity
    kubectl exec --tty -i kafka-client --namespace kafka -- bash
    PRODUCER:
        kafka-console-producer.sh \
            --broker-list kafka-0.kafka-headless.kafka.svc.cluster.local:9092,kafka-1.kafka-headless.kafka.svc.cluster.local:9092,kafka-2.kafka-headless.kafka.svc.cluster.local:9092 \
            --topic test
    CONSUMER:
        kafka-console-consumer.sh \
            --bootstrap-server kafka.kafka.svc.cluster.local:9092 \
            --topic test \
            --from-beginning

helm install --set的格式和限制

–set zookeeper.enabled=false # 禁用kafka自带zk

–set replicaCount=3 # 三个副本

–set externalZookeeper.servers=zookeeper #外部zk的地址

更多参考: Helm | Values Files

使用kafka image运行一个kafka-client pod

然后依次开启两个kafka-client bash 执行生产者和消费者的脚本进行消息的发送和消费。

kubectl run kafka-client --restart='Never' --image 10.50.10.185/kafka/docker.io/bitnami/kafka:3.2.1-debian-11-r4 --namespace kafka --command -- sleep infinity

kafka‐topics.sh ‐‐create ‐‐zookeeper 192.168.65.60:2181 ‐‐replication‐factor 1 ‐‐partitions 1 ‐‐topic test

生产者

I have no name!@kafka-client:/$ kafka-console-producer.sh \
             --broker-list kafka-0.kafka-headless.kafka.svc.cluster.local:9092,kafka-1.kafka-headless.kafka.svc.cluster.local:9092,kafka-2.kafka-headless.kafka.svc.cluster.local:9092 \
             --topic test
>test
>chottest
>asduiohfsd
>hiushdfsd
>tyufgigui

消费者

[root@master1 /nfs/data]#kubectl exec -it kafka-client -n kafka -- /bin/bash
I have no name!@kafka-client:/$ 
I have no name!@kafka-client:/$ kafka-console-consumer.sh \
             --bootstrap-server kafka.kafka.svc.cluster.local:9092 \
             --topic test \
             --from-beginning
test
chottest
asduiohfsd
hiushdfsd
tyufgigui

错误的测试，不能在kafka节点中创建producer 和consumer

I have no name!@kafka-client:/$ kafka-console-producer.sh \
>             --broker-list kafka-0.kafka-headless.default.svc.cluster.local:9092,kafka-1.kafka-headless.kafka.svc.cluster.local:9092,kafka-2.kafka-headless.kafka.svc.cluster.local:9092 \
>             --topic test
[2022-08-12 15:11:18,722] WARN Couldn't resolve server kafka-0.kafka-headless.default.svc.cluster.local:9092 from bootstrap.servers as DNS resolution failed for kafka-0.kafka-headless.default.svc.cluster.local (org.apache.kafka.clients.ClientUtils)

kafka外部如何访问

上面这个kafka集群仅仅可以K8S内部访问，如何让外部也能访问呢？你第一个想到的肯定是nodePort，同时也回忆一下其他方式，例如LB、外部DNS。

以下内容来自官方内容

Accessing Kafka brokers from outside the cluster

In order to access Kafka Brokers from outside the cluster, an additional listener and advertised listener must be configured. Additionally, a specific service per kafka pod will be created.

There are two ways of configuring external access. Using LoadBalancer services or using NodePort services.

Using LoadBalancer services

You have two alternatives to use LoadBalancer services:

Option A) Use random load balancer IPs using an initContainer that waits for the IPs to be ready and discover them automatically.

externalAccess.enabled=true
externalAccess.service.type=LoadBalancer
externalAccess.service.ports.external=9094
externalAccess.autoDiscovery.enabled=true
serviceAccount.create=true
rbac.create=true

Note: This option requires creating RBAC rules on clusters where RBAC policies are enabled.

externalAccess.enabled=true
externalAccess.service.type=LoadBalancer
externalAccess.service.ports.external=9094
externalAccess.service.loadBalancerIPs[0]='external-ip-1'
externalAccess.service.loadBalancerIPs[1]='external-ip-2'}

Note: You need to know in advance the load balancer IPs so each Kafka broker advertised listener is configured with it.

Following the aforementioned steps will also allow to connect the brokers from the outside using the cluster’s default service (when service.type is LoadBalancer or NodePort). Use the property service.externalPort to specify the port used for external connections.

Using NodePort services

You have two alternatives to use NodePort services:

Option A) Use random node ports using an initContainer that discover them automatically.

externalAccess.enabled=true
externalAccess.service.type=NodePort
externalAccess.autoDiscovery.enabled=true
serviceAccount.create=true
rbac.create=true

Note: This option requires creating RBAC rules on clusters where RBAC policies are enabled.

Option B) Manually specify the node ports:

externalAccess.enabled=true
externalAccess.service.type=NodePort
externalAccess.service.nodePorts[0]='node-port-1'
externalAccess.service.nodePorts[1]='node-port-2'

Note: You need to know in advance the node ports that will be exposed so each Kafka broker advertised listener is configured with it.

The pod will try to get the external ip of the node using curl -s https://ipinfo.io/ip unless externalAccess.service.domain or externalAccess.service.useHostIPs is provided.

Name resolution with External-DNS

You can use the following values to generate External-DNS annotations which automatically creates DNS records for each ReplicaSet pod:

externalAccess:
  service:
    annotations:
      external-dns.alpha.kubernetes.io/hostname: "{{ .targetPod }}.example.com"

helm 安装zk是保留nodeport

helm install zookeeper-cluster -n kafka --set replicaCount=3,persistence.enabled=false,service.type=NodePort /opt/helm/zookeeper
[root@master2 /opt/helm/kafka]#helm install zookeeper-cluster -n kafka --set replicaCount=3,persistence.enabled=false,service.type=NodePort /opt/helm/zookeeper
NAME: zookeeper-cluster
LAST DEPLOYED: Sat Aug 13 17:21:42 2022
NAMESPACE: kafka
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
CHART NAME: zookeeper
CHART VERSION: 10.0.7
APP VERSION: 3.8.0
** Please be patient while the chart is being deployed **
ZooKeeper can be accessed via port 2181 on the following DNS name from within your cluster:
    zookeeper-cluster.kafka.svc.cluster.local
To connect to your ZooKeeper server run the following commands:
    export POD_NAME=$(kubectl get pods --namespace kafka -l "app.kubernetes.io/name=zookeeper,app.kubernetes.io/instance=zookeeper-cluster,app.kubernetes.io/component=zookeeper" -o jsonpath="{.items[0].metadata.name}")
    kubectl exec -it $POD_NAME -- zkCli.sh
To connect to your ZooKeeper server from outside the cluster execute the following commands:
    export NODE_IP=$(kubectl get nodes --namespace kafka -o jsonpath="{.items[0].status.addresses[0].address}")
    export NODE_PORT=$(kubectl get --namespace kafka -o jsonpath="{.spec.ports[0].nodePort}" services zookeeper-cluster)

helm 安装kafka使用externalAccess

设置服务发布方式为NodePort。关闭Kafka内置的ZooKeeper，配置为使用我们刚刚搭建ZooKeeper。

helm install kafka-cluster -n kafka --set replicaCount=3,externalAccess.enabled=true,externalAccess.service.type=NodePort,externalAccess.service.port=9094,externalAccess.autoDiscovery.enabled=true,serviceAccount.create=true,zookeeper.enabled=false,externalZookeeper.servers=zookeeper-cluster:2181,rbac.create=true /opt/helm/kafka/
[root@master2 /opt/helm/kafka]#helm install kafka-cluster -n kafka --set replicaCount=3,externalAccess.enabled=true,externalAccess.service.type=NodePort,externalAccess.service.port=9094,externalAccess.autoDiscovery.enabled=true,serviceAccount.create=true,zookeeper.enabled=false,externalZookeeper.servers=zookeeper-cluster:2181,rbac.create=true /opt/helm/kafka/
NAME: kafka-cluster
LAST DEPLOYED: Sat Aug 13 17:22:01 2022
NAMESPACE: kafka
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
CHART NAME: kafka
CHART VERSION: 18.0.8
APP VERSION: 3.2.1
---------------------------------------------------------------------------------------------
 WARNING
    By specifying "serviceType=LoadBalancer" and not configuring the authentication
    you have most likely exposed the Kafka service externally without any
    authentication mechanism.
    For security reasons, we strongly suggest that you switch to "ClusterIP" or
    "NodePort". As alternative, you can also configure the Kafka authentication.
---------------------------------------------------------------------------------------------
** Please be patient while the chart is being deployed **
Kafka can be accessed by consumers via port 9092 on the following DNS name from within your cluster:
    kafka-cluster.kafka.svc.cluster.local
Each Kafka broker can be accessed by producers via port 9092 on the following DNS name(s) from within your cluster:
    kafka-cluster-0.kafka-cluster-headless.kafka.svc.cluster.local:9092
    kafka-cluster-1.kafka-cluster-headless.kafka.svc.cluster.local:9092
    kafka-cluster-2.kafka-cluster-headless.kafka.svc.cluster.local:9092
To create a pod that you can use as a Kafka client run the following commands:
    kubectl run kafka-cluster-client --restart='Never' --image 10.50.10.185/kafka/docker.io/bitnami/kafka:3.2.1-debian-11-r4 --namespace kafka --command -- sleep infinity
    kubectl exec --tty -i kafka-cluster-client --namespace kafka -- bash
    PRODUCER:
        kafka-console-producer.sh \
            --broker-list kafka-cluster-0.kafka-cluster-headless.kafka.svc.cluster.local:9092,kafka-cluster-1.kafka-cluster-headless.kafka.svc.cluster.local:9092,kafka-cluster-2.kafka-cluster-headless.kafka.svc.cluster.local:9092 \
            --topic test
    CONSUMER:
        kafka-console-consumer.sh \
            --bootstrap-server kafka-cluster.kafka.svc.cluster.local:9092 \
            --topic test \
            --from-beginning
To connect to your Kafka server from outside the cluster, follow the instructions below:
    Kafka brokers domain: You can get the external node IP from the Kafka configuration file with the following commands (Check the EXTERNAL listener)
        1. Obtain the pod name:
        kubectl get pods --namespace kafka -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka-cluster,app.kubernetes.io/component=kafka"
        2. Obtain pod configuration:
        kubectl exec -it KAFKA_POD -- cat /opt/bitnami/kafka/config/server.properties | grep advertised.listeners
    Kafka brokers port: You will have a different node port for each Kafka broker. You can get the list of configured node ports using the command below:
        echo "$(kubectl get svc --namespace kafka -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka-cluster,app.kubernetes.io/component=kafka,pod" -o jsonpath='{.items[*].spec.ports[0].nodePort}' | tr ' ' '\n')"

externalAccess的参数

externalAccess.autoDiscovery.enabled	Enable using an init container to auto-detect external IPs/ports by querying the K8s API	FALSE
externalAccess.autoDiscovery.image.registry	Init container auto-discovery image registry	docker.io
externalAccess.autoDiscovery.image.repository	Init container auto-discovery image repository	bitnami/kubectl
externalAccess.autoDiscovery.image.tag	Init container auto-discovery image tag (immutable tags are recommended)	1.24.3-debian-11-r10
externalAccess.autoDiscovery.image.pullPolicy	Init container auto-discovery image pull policy	IfNotPresent
externalAccess.autoDiscovery.image.pullSecrets	Init container auto-discovery image pull secrets	[]
externalAccess.autoDiscovery.resources.limits	The resources limits for the auto-discovery init container	{}
externalAccess.autoDiscovery.resources.requests	The requested resources for the auto-discovery init container	{}

如何解决重新安装kafka导致应用程序需要重新配置bootstrap？

什么是bootstrap？

bootstrap.servers只是用于客户端启动（bootstrap）的时候有一个可以热启动的一个连接者，一旦启动完毕客户端就应该可以得知当前集群的所有节点的信息，日后集群扩展的时候客户端也能够自动实时的得到新节点的信息，即使bootstrap.servers里面的挂掉了也应该是能正常运行的，除非节点挂掉后客户端也重启了。

所以，bootstrap.servers参数，只在启动客户端kafka连接时有用，且列表里面多个项，只要一个可用就行了。

helm安装的kafka对外暴露了三个nodeport svc.，每次重启svc，这个nodeport就会变。导致应用连接不上。如何解决这个问题呢？

固定nodePort.

安装参数中加入如下三个参数:

externalAccess.service.nodePorts[0]=‘30787’,externalAccess.service.nodePorts[1]=‘31317’,externalAccess.service.nodePorts[2]=‘30113’

helm install kafka-cluster -n kafka --set replicaCount=3,externalAccess.enabled=true,externalAccess.service.type=NodePort,externalAccess.service.nodePorts[0]='30787',externalAccess.service.nodePorts[1]='31317',externalAccess.service.nodePorts[2]='30113',externalAccess.service.port=9094,externalAccess.autoDiscovery.enabled=true,serviceAccount.create=true,zookeeper.enabled=false,externalZookeeper.servers=zookeeper-cluster:2181,rbac.create=true /opt/helm/kafka/

kafka manager 可视化

访问地址: http://10.50.10.31:30900/

kafka manager yaml文件

注意: spec.env.ZK_HOSTS 一定要是你的zk集群名字。否则会一直连接不上.

apiVersion: v1
kind: Service
metadata:
  name: kafka-manager
  namespace: kafka  
  labels:
    app: kafka-manager
spec:
  type: NodePort
  ports:
  - name: kafka
    port: 9000
    targetPort: 9000
    nodePort: 30900
  selector:
    app: kafka-manager
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: kafka-manager
  namespace: kafka  
  labels:
    app: kafka-manager
spec:
  replicas: 1
  selector:
    matchLabels:
      app: kafka-manager
  template:
    metadata:
      labels:
        app: kafka-manager
    spec:
      containers:
      - name: kafka-manager
        image: 10.50.10.185/kafka/ninesun0318/kafka-manager:1.3.3.22
        imagePullPolicy: IfNotPresent
        ports:
        - name: kafka-manager
          containerPort: 9000
          protocol: TCP
        env:
        - name: ZK_HOSTS
          value: "zookeeper-cluster:2181" # zk集群svc的名字
        livenessProbe:
          httpGet:
            path: /api/health
            port: kafka-manager
        readinessProbe:
          httpGet:
            path: /api/health
            port: kafka-manager

kafka集群压测

kafka-producer-perf-test.sh --topic pressTest --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=10.50.10.33:30787,10.50.10.34:31317,10.50.10.35:30113 acks=1
# --record-size 每条消息大小
# --throughput -1  -1代表不限流
# acks=1 代表发送给broke 至少确认leader写入log成功才算发送成功

测试日志

I have no name!@kafka-client:/$ kafka-producer-perf-test.sh --topic pressTest --num-records 1000000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=10.50.10.33:30787,10.50.10.34:31317,10.50.10.35:30113 acks=1
[2022-08-17 07:01:00,460] WARN [Producer clientId=perf-producer-client] Error while fetching metadata with correlation id 1 : {pressTest=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
132271 records sent, 26454.2 records/sec (25.83 MB/sec), 786.4 ms avg latency, 1099.0 ms max latency.
198150 records sent, 39630.0 records/sec (38.70 MB/sec), 736.4 ms avg latency, 860.0 ms max latency.
180015 records sent, 36003.0 records/sec (35.16 MB/sec), 844.1 ms avg latency, 1048.0 ms max latency.
173220 records sent, 34644.0 records/sec (33.83 MB/sec), 889.0 ms avg latency, 1096.0 ms max latency.
208410 records sent, 41682.0 records/sec (40.71 MB/sec), 719.4 ms avg latency, 832.0 ms max latency.
1000000 records sent, 35870.578951 records/sec (35.03 MB/sec), 792.42 ms avg latency, 1099.00 ms max latency, 779 ms 50th, 1007 ms 95th, 1069 ms 99th, 1097 ms 99.9th.

kafka 清除log设置

安装kafka的时候需要设定一下这几个参数，否则不能达到自动删除topic的效果.


deleteTopicEnable	Switch to enable topic deletion or not	FALSE
logRetentionBytes	A size-based retention policy for logs	_1073741824
logRetentionCheckIntervalMs	The interval at which log segments are checked to see if they can be deleted	300000
logRetentionHours	The minimum age of a log file to be eligible for deletion due to age	168

kafka-cluster-0 1/1 Running 0 4d20h

logRetentionHours 消息保留时间

deleteTopicEnable 是否自动删除，

Enables delete topic. Delete topic through the admin tool will have no effect if this config is turned off

如果没有设置 delete.topic.enable=true，则调用kafka 的delete命令无法真正将topic删除，而是显示（marked for deletion）

logRetentionCheckIntervalMs 删除检查周期

参考

https://blog.csdn.net/Hlroliu/article/details/113354423

https://artifacthub.io/packages/helm/bitnami/kafka

踩坑记

这个问题困扰了我两个多小时…

使用helm 安装zk集群，pod一直readliness 和liveness过不去，提示无法找到nc命令，很是奇怪，这可是官方给的chart，怎么可能在build镜像时没有安装探针检查的命令呢？

后来才发现使用的zk版本不一致导致的。我使用的是3.4 的镜像，但是chart使用的是3.8的。所以以后helm安装软件时一定要注意这一点，镜像版本和chart版本要一致.

10.50.10.185/kafka/docker.io/bitnami/zookeeper:3.8.0-debian-11-r24 # 探针新增了nc命令
10.50.10.185/kafka/docker.io/bitnami/zookeeper:3.4.14-debian-9-r25

另外Stack Overflow上说这个探针失败的原因可能是:

没有设置initialDelaySeconds ，导致容器一致重启。

initialDelaySeconds:10指定容器启动10s之后开始执行

Liveness探测,我们一般会根据应用启动的准备时间来设置。比如某个应用正常启动要花30秒,那么initialDelaySeconds的值就应该大于30，这样容器就不至于一直重启。

kakfa资料多久会删除？

logRetentionHours  168h

kafka log在哪里看？

persistence.mountPath /opt/bitnami/kafka/logs

Kafka集群 on k8s实战＜云原生＞

概述

安装helm

文件下载

安装helm

helm install zk

确认pod状态

kafka的zk存储结构

helm install kafka

helm install --set的格式和限制

使用kafka image运行一个kafka-client pod

生产者

消费者

kafka外部如何访问

helm 安装zk是保留nodeport

helm 安装kafka使用externalAccess

如何解决重新安装kafka导致应用程序需要重新配置bootstrap？

什么是bootstrap？

kafka manager 可视化

kafka manager yaml文件

kafka集群压测

测试日志

kafka 清除log设置

参考

踩坑记

kakfa资料多久会删除？

kafka log在哪里看？

热门文章

最新文章

相关课程

相关电子书

相关实验场景

推荐镜像