Kafka 4.0 发布，彻底移除 ZooKeeper！

大家好，我是君哥。

最近 Apache Kafka 4.0 版本发布，这是 Kafka 的一个重要里程碑。Kafka 移除 ZooKeeper 早在几年前就开始准备，4.0 版本是第一个可以完全不使用 ZooKeeper 的版本。

Kafka 4.0 默认允许使用 KRaft 模式，大大简化了集群的部署和管理，消除了集成 ZooKeeper 的复杂性。

在 KRaft 模式下，一些配置、指标和功能发生了不小的变化，今天我们来看一下都有哪些改变。

这些配置在 ZooKeeper 模式下使用，用来定义和备份密码，这个密钥用于加密加密敏感数据（如密码）。需要指定加密的算法和密钥生成方法（如 AES、RSA），并控制密钥长度和加密强度。相关参数如下：

password.encoder.secret
password.encoder.old.secret
password.encoder.keyfactory.algorithm
password.encoder.cipher.algorithm
password.encoder.key.length
password.encoder.iterations

在 KRaft 模式下，Kafka 将敏感数据存储在 records，这些敏感数据不需要加密。

Removed control.plane.listener.name

Kafka 依赖 ZooKeeper 管理元数据，但是一些内部操作（比如 Broker Controller 和普通 Broker 之间通信）依然需要 Kafka 内部控制面板来协调。

在 KRaft 模式下，Kafka 消除了对 ZooKeeper 的依赖，控制面板功能完全集成到 Kafka 内部。处理角色明确分开：brokers 处理与数据相关的请求，而 broker controllers 管理跟元数据相关的请求。controllers 使用 Raft 协议进行内部通信，操作方式与 ZooKeeper 模型不同。使用以下参数来配置控制面板监听器：

controller.listener.names
listeners
listener.security.protocol.map

使用外部注册中心，多数情况是需要配置优雅停服的。这些配置在 ZooKeeper 模式下用于定义主动停服的最大重试次数和重试回退时间。它可以降低意外 Controller 变更和数据不一致的风险。

controlled.shutdown.max.retries
controlled.shutdown.retry.backoff.ms

在 KRaft 模式下，Kafka 使用 Raft 协议来管理元数据。Broker 关闭过程跟基于仲裁模式的 ZooKeeper 模式不同。Raft 协议的 Controller 切换依赖自动领导者切换，元数据由 Controller 更新，关闭过程更加可靠和高效。

这些配置在 ZooKeeper 模式下用于自动生成 Broker id 并控制 Broker id 生成过程。

reserved.broker.max.id
broker.id.generation.enable

Kafka在 KRaft 模式下使用 node id 来标识 server。

node.id

inter.broker.protocol.version

这些配置在 ZooKeeper 模式下用于定义 Broker 之间的通信协议版本。在 KRaft 模式下，Kafka 使用 metadata.version
 来控制集群的特征级别，并且可以使用 bin/Kafka-features.sh
 脚本进行管理。

advertised.listeners

在 KRaft 模式下，要修改这些配置，就需要重启 Broker。

leader.imbalance.per.broker.percentage

这个配置只在 ZooKeeper 中使用。用于限制 ZooKeeper 中的选举领导者的频率。

zookeeper.connect
zookeeper.session.timeout.ms
zookeeper.connection.timeout.ms
zookeeper.set.acl
zookeeper.max.in.flight.requests
zookeeper.ssl.client.enable
zookeeper.clientCnxnSocket
zookeeper.ssl.keystore.location
zookeeper.ssl.keystore.password
zookeeper.ssl.keystore.type
zookeeper.ssl.truststore.location
zookeeper.ssl.truststore.password
zookeeper.ssl.truststore.type
zookeeper.ssl.protocol
zookeeper.ssl.enabled.protocols
zookeeper.ssl.cipher.suites
zookeeper.ssl.endpoint.identification.algorithm
zookeeper.ssl.crl.enable
zookeeper.ssl.ocsp.enable

下面参数监控网络处理器平均空闲率：

ControlPlaneNetworkProcessorAvgIdlePercent

下面指标监控所有处理器上断开的连接总数：

ControlPlaneExpiredConnectionsKilledCount

在 KRaft 模式下，Kafka 提供了网络处理器连接过期指标。指标如下：

NetworkProcessorAvgIdlePercent
ExpiredConnectionsKilledCount

kafka.controller:type=ControllerChannelManager,name=QueueSize
kafka.controller:type=ControllerChannelManager,name=RequestRateAndQueueTimeMs
kafka.controller:type=ControllerEventManager,name=EventQueueSize
kafka.controller:type=ControllerEventManager,name=EventQueueTimeMs
kafka.controller:type=ControllerStats,name=AutoLeaderBalanceRateAndTimeMs
kafka.controller:type=ControllerStats,name=ControlledShutdownRateAndTimeMs
kafka.controller:type=ControllerStats,name=ControllerChangeRateAndTimeMs
kafka.controller:type=ControllerStats,name=ControllerShutdownRateAndTimeMs
kafka.controller:type=ControllerStats,name=IdleRateAndTimeMs
kafka.controller:type=ControllerStats,name=IsrChangeRateAndTimeMs
kafka.controller:type=ControllerStats,name=LeaderAndIsrResponseReceivedRateAndTimeMs
kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs
kafka.controller:type=ControllerStats,name=ListPartitionReassignmentRateAndTimeMs
kafka.controller:type=ControllerStats,name=LogDirChangeRateAndTimeMs
kafka.controller:type=ControllerStats,name=ManualLeaderBalanceRateAndTimeMs
kafka.controller:type=KafkaController,name=MigratingZkBrokerCount
kafka.controller:type=ControllerStats,name=PartitionReassignmentRateAndTimeMs
kafka.controller:type=ControllerStats,name=TopicChangeRateAndTimeMs
kafka.controller:type=ControllerStats,name=TopicDeletionRateAndTimeMs
kafka.controller:type=KafkaController,name=TopicsIneligibleToDeleteCount
kafka.controller:type=ControllerStats,name=TopicUncleanLeaderElectionEnableRateAndTimeMs
kafka.controller:type=ControllerStats,name=UncleanLeaderElectionEnableRateAndTimeMs
kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec
kafka.controller:type=ControllerStats,name=UpdateFeaturesRateAndTimeMs
kafka.controller:type=ControllerStats,name=UpdateMetadataResponseReceivedRateAndTimeMs
kafka.controller:type=KafkaController,name=ActiveBrokerCount
kafka.controller:type=KafkaController,name=ActiveControllerCount
kafka.controller:type=KafkaController,name=ControllerState
kafka.controller:type=KafkaController,name=FencedBrokerCount
kafka.controller:type=KafkaController,name=GlobalPartitionCount
kafka.controller:type=KafkaController,name=GlobalTopicCount
kafka.controller:type=KafkaController,name=OfflinePartitionsCount
kafka.controller:type=KafkaController,name=PreferredReplicaImbalanceCount
kafka.controller:type=KafkaController,name=ReplicasIneligibleToDeleteCount
kafka.controller:type=KafkaController,name=ReplicasToDeleteCount
kafka.controller:type=KafkaController,name=TopicsToDeleteCount
kafka.controller:type=KafkaController,name=ZkMigrationState
kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=ElectLeader
kafka.server:type=DelayedOperationPurgatory,name=PurgatorySize,delayedOperation=topic
kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=ElectLeader
kafka.server:type=DelayedOperationPurgatory,name=NumDelayedOperations,delayedOperation=topic
kafka.server:type=SessionExpireListener,name=SessionState
kafka.server:type=SessionExpireListener,name=ZooKeeperAuthFailuresPerSec
kafka.server:type=SessionExpireListener,name=ZooKeeperDisconnectsPerSec
kafka.server:type=SessionExpireListener,name=ZooKeeperExpiresPerSec
kafka.server:type=SessionExpireListener,name=ZooKeeperReadOnlyConnectsPerSec
kafka.server:type=SessionExpireListener,name=ZooKeeperSaslAuthenticationsPerSec
kafka.server:type=SessionExpireListener,name=ZooKeeperSyncConnectsPerSec
kafka.server:type=ZooKeeperClientMetrics,name=ZooKeeperRequestLatencyMs