Kafka3.x集群搭建(Kraft模式)
https://kafka.apache.org/downloads
kafka_2.12-3.1.0.tgz
前期准备
准备3台服务器
| HOSTNAME | IP |
|---|
| kraft1 | 192.168.25.101 | | kraft2 | 192.168.25.102 | | kraft3 | 192.168.25.103 |
分别修改三台机器的hostname
#192.168.25.101上执行
hostnamectl set-hostname kraft1
修改/etc/hosts文件（三台机器）
/etc/hosts << EOF
192.168.25.101 kraft1
192.168.25.102 kraft2
192.168.25.103 kraft3
EOF
安装JDK
下载安装包并上传到服务器
解压：
tar zxf kafka_2.12-3.1.0.tgz
安装步骤
-
- 修改kraft/server.properties
kraft1 # The node id associated with this instance's roles
node.id=1
# The connect string for the controller quorum
controller.quorum.voters=1@192.168.25.101:9093,2@192.168.25.102:9093,3@192.168.25.103:9093
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
kraft2 # The node id associated with this instance's roles
node.id=2
# The connect string for the controller quorum
controller.quorum.voters=1@192.168.25.101:9093,2@192.168.25.102:9093,3@192.168.25.103:9093
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
kraft3 # The node id associated with this instance's roles
node.id=3
# The connect string for the controller quorum
controller.quorum.voters=1@192.168.25.101:9093,2@192.168.25.102:9093,3@192.168.25.103:9093
############################# Socket Server Settings #############################
# The address the socket server listens on. It will get the value returned from
# java.net.InetAddress.getCanonicalHostName() if not configured.
# FORMAT:
# listeners = listener_name://host_name:port
# EXAMPLE:
# listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
-
- 生成集群uuid
a. 在 kraft1 执行 ./bin/kafka-storage.sh random-uuid
RYilCnKJQFWPP1ftjzSdAA b. 三台都执行如下命令 ./bin/kafka-storage.sh format -t RYilCnKJQFWPP1ftjzSdAA -c ./config/kraft/server.properties
c. 查看配置是否生效 cat /tmp/kraft-combined-logs/meta.properties
cluster.id=RYilCnKJQFWPP1ftjzSdAA
version=1
node.id=1 -
- 启动节点服务
./bin/kafka-server-start.sh ./config/kraft/server.properties
启动日志 [2022-03-15 09:43:36,701] INFO [SocketServer listenerType=BROKER, nodeId=1] Starting socket server acceptors and processors (kafka.network.SocketServer)
[2022-03-15 09:43:36,710] INFO [Transaction Marker Channel Manager 1]: Starting (kafka.coordinator.transaction.TransactionMarkerChannelManager)
[2022-03-15 09:43:36,710] INFO [LogDirFailureHandler]: Starting (kafka.server.ReplicaManager$LogDirFailureHandler)
[2022-03-15 09:43:36,731] INFO [SocketServer listenerType=BROKER, nodeId=1] Started data-plane acceptor and processor(s) for endpoint : ListenerName(PLAINTEXT) (kafka.network.SocketServer)
[2022-03-15 09:43:36,731] INFO [SocketServer listenerType=BROKER, nodeId=1] Started socket server acceptors and processors (kafka.network.SocketServer)
[2022-03-15 09:43:36,731] INFO [BrokerServer id=1] Transition from STARTING to STARTED (kafka.server.BrokerServer)
[2022-03-15 09:43:36,732] INFO Kafka version: 3.1.0 (org.apache.kafka.common.utils.AppInfoParser)
[2022-03-15 09:43:36,732] INFO Kafka commitId: 37edeed0777bacb3 (org.apache.kafka.common.utils.AppInfoParser)
[2022-03-15 09:43:36,732] INFO Kafka startTimeMs: 1647308616731 (org.apache.kafka.common.utils.AppInfoParser)
[2022-03-15 09:43:36,733] INFO Kafka Server started (kafka.server.KafkaRaftServer)
[2022-03-15 09:43:36,746] INFO [BrokerLifecycleManager id=1] The broker is in RECOVERY. (kafka.server.BrokerLifecycleManager)
[2022-03-15 09:43:36,886] INFO [Controller 1] Unfenced broker: UnfenceBrokerRecord(id=2, epoch=11) (org.apache.kafka.controller.ClusterControlManager)
[2022-03-15 09:43:38,039] INFO [Controller 1] Re-registered broker id 3: RegisterBrokerRecord(brokerId=3, incarnationId=IUgThYFITFmL6WQA6dWO1Q, brokerEpoch=15, endPoints=[BrokerEndpoint(name='PLAINTEXT', host='localhost', port=9092, securityProtocol=0)], features=[], rack=null, fenced=true) (org.apache.kafka.controller.ClusterControlManager)
[2022-03-15 09:43:38,748] INFO [BrokerLifecycleManager id=1] The broker is in RECOVERY. (kafka.server.BrokerLifecycleManager)
[2022-03-15 09:43:39,142] INFO [Controller 1] Unfenced broker: UnfenceBrokerRecord(id=3, epoch=15) (org.apache.kafka.controller.ClusterControlManager)
[2022-03-15 09:43:40,800] INFO [BrokerLifecycleManager id=1] The broker has been unfenced. Transitioning from RECOVERY to RUNNING. (kafka.server.BrokerLifecycleManager)
[2022-03-15 09:43:41,300] INFO [Controller 1] Unfenced broker: UnfenceBrokerRecord(id=1, epoch=13) (org.apache.kafka.controller.ClusterControlManager)
执行 bin/kafka-topics.sh --create --replication-factor 1 --partitions 1 --topic test --bootstrap-server localhost:9092
如果执行失败 bin/kafka-topics.sh --create --replication-factor 1 --partitions 1 --topic test --bootstrap-server 192.168.25.101:9092
OpenJDK 64-Bit Server VM warning: If the number of processors is expected to increase from one, then you should configure the number of parallel GC threads appropriately using -XX:ParallelGCThreads=N
[2022-03-15 09:48:04,100] WARN [AdminClient clientId=adminclient-1] Connection to node 3 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
[2022-03-15 09:48:04,102] WARN [AdminClient clientId=adminclient-1] Connection to node 2 (localhost/127.0.0.1:9092) could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)
检查配置文件中listeners=PLAINTEXT://:9092,CONTROLLER://:9093 在其他节点查询 bin/kafka-topics.sh --list --bootstrap-server localhost:9092
安装完成
|