1
1)三台启动 Kafka
2
root@Elk01 ~# source /etc/profile.d/kafka.sh && kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties
3
# 三台都执行 ⚠️
4
root@Elk01 ~# systemctl status zookeeper.service
5
● zookeeper.service - Apache Zookeeper
6
     Loaded: loaded (/usr/lib/systemd/system/zookeeper.service; enabled; preset: enabled)
7
     Active: active (running) since Wed 2026-08-05 19:45:03 CST; 36min ago
8
# 我这里是因为有zookeeper服务，开机自启动！
9

10
2)验证端口
11
root@Elk01 ~# ss -lnt | grep -E "9092|2181|9200|9300"
12
LISTEN 0      50                     *:9092             *:*
13
LISTEN 0      50                     *:2181             *:*
14
LISTEN 0      4096                   *:9200             *:*
15
LISTEN 0      4096                   *:9300             *:*
16
✅️ Kafka + ZK + ES 全部在线

1
1)创建 topic（5 分区 2 副本）
2
root@Elk01 ~# kafka-topics.sh --bootstrap-server 10.0.0.6:9092 --topic elfk-kafka --create --partitions 5 --replication-factor 2
3
Created topic elfk-kafka.
4
root@Elk01 ~# kafka-topics.sh --bootstrap-server 10.0.0.6:9092 --topic elfk-kafka --describe
5
Topic: elfk-kafka  PartitionCount: 5  ReplicationFactor: 2  Configs:
6
  Topic: elfk-kafka  Partition: 0  Leader: 3  Replicas: 3,2  Isr: 3,2
7
  Topic: elfk-kafka  Partition: 1  Leader: 1  Replicas: 1,3  Isr: 1,3
8
  Topic: elfk-kafka  Partition: 2  Leader: 2  Replicas: 2,1  Isr: 2,1
9
  Topic: elfk-kafka  Partition: 3  Leader: 3  Replicas: 3,1  Isr: 3,1
10
  Topic: elfk-kafka  Partition: 4  Leader: 1  Replicas: 1,2  Isr: 1,2
11

12
2)编写 Filebeat 配置（Elk02 上）
13
root@Elk02 ~# vim /etc/filebeat/config/tcp-to-kafka.yaml
14
filebeat.inputs:
15
- type: tcp
16
  host: "0.0.0.0:9000"
17

18
output.kafka:
19
  hosts: ["10.0.0.6:9092", "10.0.0.7:9092", "10.0.0.8:9092"]
20
  topic: elfk-kafka
21

22
3)启动 Filebeat
23
root@Elk02 ~# killall -9 filebeat 2>/dev/null
24
root@Elk02 ~# nohup filebeat -e -c /etc/filebeat/config/tcp-to-kafka.yaml &> /tmp/fb-kafka.log &
25
root@Elk02 ~# tail -1 /tmp/fb-kafka.log
26
Started listening for TCP connection  {"address": "0.0.0.0:9000"}
27

28
4)发送测试数据（从 Elk01 发给 Elk02 的 9000 端口）
29
root@Elk01 ~# echo www.kpyun.com | nc -w 1 10.0.0.7 9000
30

31
5)Kafka 验证数据已收到
32
oot@Elk01 ~# kafka-console-consumer.sh --bootstrap-server 10.0.0.7:9092 --topic elfk-kafka --from-beginning | jq
33
{
34
  "@timestamp": "2026-08-05T12:29:58.386Z",
35
  "@metadata": {
36
    "beat": "filebeat",
37
    "type": "_doc",
38
    "version": "7.17.29"
39
  },
40
  "ecs": {
41
    "version": "1.12.0"
42
  },
43
  "host": {
44
    "name": "Elk02"
45
  },
46
  "agent": {
47
    "name": "Elk02",
48
    "type": "filebeat",
49
    "version": "7.17.29",
50
    "hostname": "Elk02",
51
    "ephemeral_id": "a64f2ce4-80fe-434d-bfd8-96158b89d572",
52
    "id": "0faa14ff-9e2b-4b69-8840-5e8cae268f15"
53
  },
54
  "message": "www.kpyun.com",
55
  "log": {
56
    "source": {
57
      "address": "10.0.0.6:35930"
58
    }
59
  },
60
  "input": {
61
    "type": "tcp"
62
  }
63
}
64
Processed a total of 1 messages
65
✅️ Filebeat → Kafka 链路通了

1
1)创建 Logstash 专用 api-key
2
root@Elk01 ~# curl -s -u elastic:passwd -k -X POST 'https://10.0.0.6:9200/_security/api_key' \
3
    -H 'Content-Type: application/json' \
4
    -d '{"name":"elfk-lg","role_descriptors":{"logstash_writer":{"cluster":["monitor","manage_index_templates","manage_ilm"],"index":[{"names":["kpyun-es-apikey-haha*"],"privileges":["create_index","create_doc","index","read"]}]}}}' | jq
5
{
6
  "id": "MQj50Z8BjME-gAjitirQ",
7
  "name": "elfk-lg",
8
  "api_key": "93WakTr4Ruisga8oGRwSBQ",
9
  "encoded": "TVFqNTBaOEJqTUUtZ0FqaXRpclE6OTNXYWtUcjRSdWlzZ2E4b0dSd1NCUQ=="
10
}
11
root@Elk01 ~# echo 'TVFqNTBaOEJqTUUtZ0FqaXRpclE6OTNXYWtUcjRSdWlzZ2E4b0dSd1NCUQ==' | base64 -d ;echo
12
MQj50Z8BjME-gAjitirQ:93WakTr4Ruisga8oGRwSBQ
13

14
2)编写 Logstash 配置（Elk03 上）
15
root@Elk03 ~# vim /etc/logstash/conf.d/kafka-to-es.conf
16
input {
17
  kafka {
18
     bootstrap_servers => "10.0.0.6:9092,10.0.0.7:9092,10.0.0.8:9092"
19
     topics => "elfk-kafka"
20
     group_id => "linux-001"
21
     # 也可以通过更改组id重新获取数据
22
     auto_offset_reset => "earliest"
23
     # topic 从头开始消费
24
  }
25
}
26

27
filter {
28
  json {
29
    source => "message"
30
  }
31
  mutate {
32
     remove_field => [ "@version","agent","log","ecs","tags","input" ]
33
  }
34
}
35

36
output {
37
  elasticsearch {
38
    hosts => ["https://10.0.0.6:9200","https://10.0.0.7:9200","https://10.0.0.8:9200"]
39
    index => "kpyun-es-apikey-haha-01"
40
    api_key => "MQj50Z8BjME-gAjitirQ:93WakTr4Ruisga8oGRwSBQ"
41
    ssl => true
42
    ssl_certificate_verification => false
43
  }
44
}
45

46
3)手动创建索引（我们关掉了自动创建）
47
root@Elk01 ~# curl -s -u elastic:passwd -k -X PUT "https://10.0.0.6:9200/kpyun-es-apikey-haha-01" -H "Content-Type: application/json"
48
{"acknowledged":true,"shards_acknowledged":true,"index":"kpyun-es-apikey-haha-2026-08-06"}
49

50
4)启动 Logstash
51
root@Elk03 ~# pkill -f logstash; nohup /usr/share/logstash/bin/logstash -rf /etc/logstash/conf.d/kafka-to-es.conf &> /tmp/ls-kafka.log &
52
root@Elk03 ~# tail -f /tmp/ls-kafka.log
53
[ERROR] 2026-08-06 09:53:16.981 [pool-7-thread-1] jvm - Unknown garbage collector name {:name=>"G1 Concurrent GC"}
54
# 本质是 Logstash 7.x 的 Ruby 层日志和 Java 层日志是两套体系。jvm.rb 用 logger.error() 直接输出，它不受 log4j2 控制
55
✅ 改源码 —— 把 jvm.rb 第 23 行 logger.error 改成 logger.debug：
56
root@Elk03 ~# sed -i 's/logger.error(\"Unknown garbage collector name\"/logger.debug(\"Unknown garbage collector name\"/'  \
57
    /usr/share/logstash/logstash-core/lib/logstash/instrument/periodic_poller/jvm.rb
58
root@Elk03 ~# pkill -f logstash; nohup /usr/share/logstash/bin/logstash -rf /etc/logstash/conf.d/kafka-to-es.conf &> /tmp/ls-kafka.log &
59
# 重新启动
60

61
5)发送测试数据验证全链路
62
root@Elk01 ~# echo "kpyun-elfk-test" | nc -w 1 10.0.0.7 9000
63

64
6)ES 查询确认数据落地
65
root@Elk01 ~# curl -s -u elastic:passwd -k "https://10.0.0.6:9200/kpyun-es-apikey-haha-01/_search?size=2" | jq '.hits.hits[1]._source'
66
`拉了两条数据，查看第二条数据`
67
{
68
  "message": "kpyun-elfk-test",
69
  "host": {
70
    "name": "Elk02"
71
  },
72
  "@timestamp": "2026-08-06T01:47:03.262Z"
73
}
74
✅️ Filebeat → Kafka → Logstash → ES 全链路贯通！

1
1)下载 go1.25.0
2
jiuzhao@Ubuntu ~$ wget -P /home/jiuzhao/下载 https://golang.google.cn/dl/go1.25.0.linux-amd64.tar.gz
3
# 官方源被墙，用 google.cn 镜像
4

5
2)拷贝并安装到 Elk02
6
jiuzhao@Ubuntu ~$ scp /home/jiuzhao/下载/go1.25.0.linux-amd64.tar.gz Elk02:/tmp/
7
root@Elk02 ~# tar xf /tmp/go1.25.0.linux-amd64.tar.gz -C /usr/local/
8

9
3)配置环境变量
10
root@Elk02 ~# mkdir -p /jiuzhao/gopath
11
root@Elk02 ~# vim /etc/profile.d/go.sh
12
#!/bin/bash
13
export GOROOT=/usr/local/go/
14
export GOPROXY=https://goproxy.cn,direct
15
export GOPATH=/jiuzhao/gopath
16
export PATH=$PATH:$GOROOT/bin:$GOPATH
17
root@Elk02 ~# source /etc/profile.d/go.sh && go version
18
go version go1.25.0 linux/amd64

1
3)运行生成
2
root@Elk02 devops# go run main.go
3
# 近 3 年均匀分布
4
开始生成1000000条Nginx访问日志（时间范围：2023-08-05 16:47:33 至 2026-08-05 16:47:33）...
5
日志生成完成！
6
总记录数：1000000条
7
耗时：2.182946241s
8
文件大小：196.01 MB
9
日志文件路径：nginx_access.log
10

11
4)验证日志格式
12
root@Elk02 devops# wc -l nginx_access.log
13
1000000 nginx_access.log
14
root@Elk02 devops# tail -1 nginx_access.log
15
211.98.210.150 - - [07/May/2025:17:13:31 +0800] "DELETE /admin/login HTTP/2.0" 201 409 "https://github.com" "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) CriOS/120.0.6099.119 Mobile/15E148 Safari/604.1"

1
1)创建 topic
2
root@Elk01 ~# kafka-topics.sh --bootstrap-server 10.0.0.6:9092 --topic elfk-kafka-nginx --create --partitions 5 --replication-factor 2
3
Created topic elfk-kafka-nginx.
4
root@Elk02 devops# kafka-topics.sh --bootstrap-server 10.0.0.6:9092 --topic elfk-kafka-nginx --describe
5
Topic: elfk-kafka-nginx  PartitionCount: 5  ReplicationFactor: 2  Configs:
6
  Topic: elfk-kafka-nginx  Partition: 0  Leader: 2  Replicas: 2,3  Isr: 2,3
7
  Topic: elfk-kafka-nginx  Partition: 1  Leader: 3  Replicas: 3,1  Isr: 3,1
8
  Topic: elfk-kafka-nginx  Partition: 2  Leader: 1  Replicas: 1,2  Isr: 1,2
9
  Topic: elfk-kafka-nginx  Partition: 3  Leader: 2  Replicas: 2,1  Isr: 2,1
10
  Topic: elfk-kafka-nginx  Partition: 4  Leader: 3  Replicas: 3,2  Isr: 3,2
11

12
2)Filebeat 配置（filestream 读取日志文件）
13
root@Elk02 ~# vim /etc/filebeat/config/nginx-to-kafka.yaml
14
filebeat.inputs:
15
- type: filestream
16
  paths:
17
    - /jiuzhao/code/devops/nginx_access.log
18

19
output.kafka:
20
  hosts: ["10.0.0.6:9092", "10.0.0.7:9092", "10.0.0.8:9092"]
21
  topic: elfk-kafka-nginx
22

23
root@Elk02 ~# killall -9 filebeat
24
root@Elk02 ~# nohup filebeat -e -c /etc/filebeat/config/nginx-to-kafka.yaml &> /tmp/fb-nginx.log &
25
# 30 秒内 100w 条全部送入 Kafka ✅️
26

27
3)创建新的 api-key（给 nginx 索引写权限）
28
root@Elk01 ~# curl -s -u elastic:passwd -k -X POST 'https://10.0.0.6:9200/_security/api_key' \
29
    -H 'Content-Type: application/json' \
30
    -d '{"name":"elfk-nginx","role_descriptors":{"logstash_writer":{"cluster":["monitor","manage_index_templates","manage_ilm"],"index":[{"names":["kpyun-kafka-nginx*"],"privileges":["create_index","create_doc","index","read"]}]}}}' | jq
31
{
32
  "id": "8sPK1Z8BlR5ZL1RhGdS4",
33
  "name": "elfk-nginx",
34
  "api_key": "yUDrmCLvQouUYTV24OYIkw",
35
  "encoded": "OHNQSzFaOEJsUjVaTDFSaEdkUzQ6eVVEcm1DTHZRb3VVWVRWMjRPWUlrdw=="
36
}
37
root@Elk01 ~# echo 'OHNQSzFaOEJsUjVaTDFSaEdkUzQ6eVVEcm1DTHZRb3VVWVRWMjRPWUlrdw==' | base64 -d ;echo
38
8sPK1Z8BlR5ZL1RhGdS4:yUDrmCLvQouUYTV24OYIkw
39

40
4)Logstash 配置（grok + date + useragent + geoip）
41
root@Elk03 ~# vim /etc/logstash/conf.d/kafka-to-es.conf
42
input {
43
  kafka {
44
     bootstrap_servers => "10.0.0.6:9092,10.0.0.7:9092,10.0.0.8:9092"
45
     topics => "elfk-kafka-nginx"
46
     group_id => "linux-006"
47
     auto_offset_reset => "earliest"
48
  }
49
}
50

51
filter {
52
  json {
53
    source => "message"
54
  }
55
  mutate {
56
     remove_field => [ "@version","agent","log","ecs","tags","input" ]
57
  }
58
  grok {
59
     match => { "message" => "%{HTTPD_COMBINEDLOG}" }
60
     # 基于正则套模板
61
  }
62
  date {
63
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
64
    target => "@timestamp"
65
  }
66
  useragent {
67
     source => "agent"
68
     target => "kpyun-agent"
69
  }
70
  geoip {
71
     source => "clientip"
72
     database => "/root/GeoLite2-City_20250311/GeoLite2-City.mmdb"
73
     default_database_type => "City"
74
  }
75
}
76

77
output {
78

79
#  stdout {
80
#     codec => "rubydebug"
81
#  }
82

83
  elasticsearch {
84
    hosts => ["https://10.0.0.6:9200","https://10.0.0.7:9200","https://10.0.0.8:9200"]
85
    index => "kpyun-kafka-nginx-test"
86
    api_key => "8sPK1Z8BlR5ZL1RhGdS4:yUDrmCLvQouUYTV24OYIkw"
87
    ssl => true
88
    ssl_certificate_verification => false
89
  }
90
}

1
6)启动 Logstash
2
root@Elk03 ~# nohup /usr/share/logstash/bin/logstash -rf /etc/logstash/conf.d/kafka-to-es.conf &> /tmp/ls-nginx.log &
3

4
7)验证解析结果（100w 条全部写入）
5
root@Elk03 ~# curl -s -u elastic:passwd -k "https://10.0.0.6:9200/kpyun-kafka-nginx-test/_count"
6
{"count":1000000,"_shards":{"total":7,"successful":7,"skipped":0,"failed":0}}
7

8
8)查看样例文档
9
root@Elk03 ~# curl -s -u elastic:passwd -k "https://10.0.0.6:9200/kpyun-kafka-nginx-test/_search?size=1" | jq '.hits.hits[0]._source'
10
{
11
  "timestamp": "10/Aug/2024:18:18:45 +0800",
12
  "httpversion": "1.0",
13
  "message": "210.175.8.194 - - [10/Aug/2024:18:18:45 +0800] \"GET /index.html HTTP/1.0\" 404 3927 \"https://www.baidu.com\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36\"",
14
  "verb": "GET",
15
  "response": "404",
16
  "ident": "-",
17
  "geoip": {
18
    "continent_code": "AS",
19
    "country_code2": "JP",
20
    "timezone": "Asia/Tokyo",
21
    "latitude": 35.69,
22
    "longitude": 139.69,
23
    "country_code3": "JP",
24
    "location": {
25
      "lon": 139.69,
26
      "lat": 35.69
27
    },
28
    "country_name": "Japan",
29
    "ip": "210.175.8.194"
30
  },
31
  "kpyun-agent": {
32
    "name": "Chrome",
33
    "major": "119",
34
    "version": "119.0.0.0",
35
    "minor": "0",
36
    "os": "Mac OS X",
37
    "os_version": "10.15.7",
38
    "os_name": "Mac OS X",
39
    "patch": "0",
40
    "os_minor": "15",
41
    "device": "Mac",
42
    "os_full": "Mac OS X 10.15.7",
43
    "os_major": "10",
44
    "os_patch": "7"
45
  },
46
  "@timestamp": "2024-08-10T10:18:45.000Z",
47
  "host": {
48
    "name": "Elk02"
49
  },
50
  "clientip": "210.175.8.194",
51
  "request": "/index.html",
52
  "referrer": "\"https://www.baidu.com\"",
53
  "bytes": "3927",
54
  "auth": "-",
55
  "agent": "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36\""
56
}
57
✅️ grok 解析正确、geoip 定位到日本、useragent 识别出 Chrome 119 + Mac OS X 10.15.7！

1
1)恢复快照（ESnew 三台）
2
jiuzhao@Ubuntu ~$ for i in ESnew-0{1..3} ;do virsh snapshot-revert $i init ;done
3
jiuzhao@Ubuntu ~$ for i in ESnew-0{1..3} ;do virsh start $i ;done
4

5
2)安装 JDK8 & kafka（ESnew 上是干净系统，没有 JDK）
6
jiuzhao@Ubuntu ~$ scp jdk8-temurin.tar.gz ESnew-01:/tmp/
7
jiuzhao@Ubuntu ~$ scp kafka_2.13-3.9.2.tgz ESnew-01:/tmp/
8
root@ESnew-01 ~# tar xzf /tmp/jdk8-temurin.tar.gz -C /usr/local/
9
root@ESnew-01 ~# tar xf /tmp/kafka_2.13-3.9.2.tgz -C /usr/local/
10

11
3)配置环境变量
12
root@ESnew-01 ~# vim /etc/profile.d/kafka.sh
13
#!/bin/bash
14
export KAFKA_HOME=/usr/local/kafka_2.13-3.9.2
15
export JAVA_HOME=/usr/local/jdk8u502-b07
16
export PATH=$PATH:$KAFKA_HOME/bin:$JAVA_HOME/bin
17

18
4)预调 JVM 堆（2G 机器避免 OOM）
19
root@ESnew-01 ~# sed -i 's/KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"/KAFKA_HEAP_OPTS="-Xmx256m -Xms256m"/' \
20
  $KAFKA_HOME/bin/kafka-server-start.sh
21

22
5)修改 server.properties — KRaft 核心配置
23
root@ESnew-01 ~# source /etc/profile.d/kafka.sh
24
root@ESnew-01 ~# vim $KAFKA_HOME/config/server.properties
25
broker.id=1                              # ESnew-02: 2, ESnew-03: 3
26
advertised.listeners=PLAINTEXT://10.0.0.9:9092
27

28
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
29
# 👆 两个端口：9092 给客户端，9093 给控制器通信
30

31
log.dirs=/var/lib/kafka
32

33
# ZK 配置必须注释掉！
34
# zookeeper.connect=localhost:2181
35
# zookeeper.connection.timeout.ms=18000
36

37
# KRaft 核心三项配置：（在末尾追加）
38
process.roles=broker,controller
39
# 节点角色（broker + controller 复用）
40
controller.listener.names=CONTROLLER
41
# controller 通信用的 listener 名称
42
controller.quorum.voters=1@10.0.0.9:9093,2@10.0.0.10:9093,3@10.0.0.11:9093
43
# 👆 quorum 投票者列表——KRaft 模式"不需要 ZK"的底气就在这里

1
6)各节点创建数据目录 + 添加 hosts 解析
2
root@ESnew-01 ~# mkdir -p /var/lib/kafka
3
root@ESnew-01 ~# cat >> /etc/hosts <<EOF
4
10.0.0.9  ESnew-01
5
10.0.0.10 ESnew-02
6
10.0.0.11 ESnew-03
7
EOF
8
# 三台都加 hosts，KRaft 需要主机名解析
9

10
7)拷贝jdk、kafka、环境变量、hosts
11
root@ESnew-01 ~# scp -r /usr/local/{jdk8u502-b07,kafka_2.13-3.9.2} ESnew-02:/usr/local
12
root@ESnew-01 ~# scp -r /usr/local/{jdk8u502-b07,kafka_2.13-3.9.2} ESnew-03:/usr/local
13
root@ESnew-01 ~# scp /etc/profile.d/kafka.sh ESnew-02:/etc/profile.d/
14
root@ESnew-01 ~# scp /etc/profile.d/kafka.sh ESnew-03:/etc/profile.d/
15
root@ESnew-01 ~# scp /etc/hosts ESnew-02:/etc/
16
root@ESnew-01 ~# scp /etc/hosts ESnew-03:/etc/
17

18
8)更改主配置 & 创建数据目录
19
root@ESnew-02 ~# source /etc/profile.d/kafka.sh && vim $KAFKA_HOME/config/server.properties
20
broker.id=2
21
# ESnew-02: 2, ESnew-03: 3
22
advertised.listeners=PLAINTEXT://10.0.0.10:9092
23
....
24
root@ESnew-03 ~# source /etc/profile.d/kafka.sh && vim $KAFKA_HOME/config/server.properties
25
broker.id=3
26
# ESnew-02: 2, ESnew-03: 3
27
advertised.listeners=PLAINTEXT://10.0.0.11:9092
28
root@ESnew-02 ~# mkdir -p /var/lib/kafka
29
root@ESnew-03 ~# mkdir -p /var/lib/kafka
30

31
9)生成集群 UUID
32
root@ESnew-01 ~# source /etc/profile.d/kafka.sh
33
root@ESnew-01 ~# kafka-storage.sh random-uuid
34
xsUe7WA7SXWRxOaDM92JmQ
35

36
10)三台初始化（格式化元数据目录）
37
root@ESnew-01 ~# kafka-storage.sh format -t xsUe7WA7SXWRxOaDM92JmQ -c $KAFKA_HOME/config/server.properties
38
`把集群 UUID 和 node.id 写进本地，有了"身份信息"后再启动 Kafka`
39
Formatting metadata directory /var/lib/kafka with metadata.version 3.9-IV0.
40
# 三台同时执行同一个 UUID
41

42
root@ESnew-01 ~# cat /var/lib/kafka/meta.properties
43
#Thu Aug 06 21:09:30 CST 2026
44
cluster.id=xsUe7WA7SXWRxOaDM92JmQ    # 同一集群的 UUID 统一
45
version=1
46
directory.id=qqnY7cCi518hXOBd4EfjcQ    # 每个节点的唯一目录 ID
47
node.id=1                # 节点 ID
48
# ESnew-02: node.id=2, ESnew-03: node.id=3

1
1)创建 topic
2
root@ESnew-01 ~# kafka-topics.sh --bootstrap-server 10.0.0.9:9092 --topic kpyun-linux --create
3
Created topic kpyun-linux.
4

5
root@ESnew-01 ~# kafka-topics.sh --bootstrap-server 10.0.0.9:9092 --topic kpyun-linux --describe
6
Topic: kpyun-linux  PartitionCount: 1  ReplicationFactor: 1  Configs:
7
  Topic: kpyun-linux  Partition: 0  Leader: 2  Replicas: 2  Isr: 2
8
✅️ KRaft 模式 topic 正常
9

10
2)生产者写数据
11
root@ESnew-01 ~# echo "hello-kraft-cluster" | kafka-console-producer.sh --bootstrap-server 10.0.0.9:9092 --topic kpyun-linux
12

13
3)消费者读数据（跨节点消费）
14
root@ESnew-02 ~# kafka-console-consumer.sh --bootstrap-server 10.0.0.10:9092 --topic kpyun-linux --from-beginning
15
hello-kraft-cluster
16
✅️ KRaft 集群跨节点消费正常，无需 ZK！