Kafka中的动态配置源码分析

kafka管控推荐使用滴滴开源的 Kafka运维管控平台更符合国人的操作习惯 ,
更强大的管控能力 ,更高效的问题定位能力、更便捷的集群运维能力、更专业的资源治理、更友好的运维生态

动态配置的使用,请看【kafka运维】Kafka全网最全最详细运维手册!!! 的动态配置部分

源码分析

Broker启动加载动态配置

KafkaServer.startup

启动加载动态配置总流程

1. 动态配置初始化

1	config.dynamicConfig.initialize(zkClient)

构造当前配置文件 currentConfig，然后从zk中获取节点 /config/brokers/<default>信息,然后更新配置updateDefaultConfig; (动态默认配置覆盖静态配置)
从节点/config/brokers/{当前BrokerId}获取配置, 如果配置中有ConfigType=PASSWORD的配置(例如ssl.keystore.password)存在,接着判断是否存在password.encoder.old.secret 配置,（这个配置是用来加解密ConfigType=PASSWORD的旧的秘钥）,尝试用旧秘钥解密秘钥; 然后将这些配置重新加密回写入/config/brokers/{当前BrokerId} ; 然后返回配置（这里主要是动态配置里面有密码类型配置的时候需要做一次解密加密处理）
将上面得到的配置(password类型修改之后) 更新内存总的配置；优先级静态配置<动态默认配置<指定动态配置

2. 注册可变更配置监听器

如果有对应的配置变更了,那么相应的监听器就会收到通知去修改自己相应的配置;

1
2
3


config.dynamicConfig.addReconfigurables(this)

DynamicBrokerConfig.addReconfigurables

// .........
  def addReconfigurables(kafkaServer: KafkaServer): Unit = {
    kafkaServer.authorizer match {
      case Some(authz: Reconfigurable) => addReconfigurable(authz)
      case _ =>
    }
    addReconfigurable(new DynamicMetricsReporters(kafkaConfig.brokerId, kafkaServer))
    addReconfigurable(new DynamicClientQuotaCallback(kafkaConfig.brokerId, kafkaServer))

    addBrokerReconfigurable(new DynamicThreadPool(kafkaServer))
    if (kafkaServer.logManager.cleaner != null)
      addBrokerReconfigurable(kafkaServer.logManager.cleaner)
    addBrokerReconfigurable(new DynamicLogConfig(kafkaServer.logManager, kafkaServer))
    addBrokerReconfigurable(new DynamicListenerConfig(kafkaServer))
    addBrokerReconfigurable(kafkaServer.socketServer)
  }

3. 动态配置启动监听

1
2
3

// Create the config manager. start listening to notifications
dynamicConfigManager = new DynamicConfigManager(zkClient, dynamicConfigHandlers)
dynamicConfigManager.startup()

注册节点处理器change-notification-/config/changes = stateChangeHandler
注册节点处理器/config/changes = zNodeChildChangeHandler
获取/config/changes 所有子节点看看有哪些变更
遍历所有节点并截取节点的编号, 判断一下是不是大于上一次执行过变更的节点ID lastExecutedChange(启动的时候是-1)
上个条件满足的话,则执行通知操作；不同entity执行的操作不一样,具体请看下面每个类型
更新lastExecutedChange
清除过期的通知节点, 默认过期时间15 * 60 * 1000(15分钟) 就是删除/config/changes /下面的过期节点

加载Topic动态配置

TopicConfigHandler.processConfigChanges

获取节点的data数据, 如果获取到了则执行通知流程notificationHandler.processNotification(d),处理器是ConfigChangedNotificationHandler; 它先解析节点的json数据,根据版本信息不同调用不同的处理方法; 下面是version=2的处理方式;
根据json数据可以得到 entityType 和entityName; 那么久可以去对应的zk数据里面getData获取数据；并且将获取到的数据Decode成Properties对象entityConfig;
将key为下图中的属性隐藏掉; 替换成value: [hidden]
调用EntityHandler; 这里是TopicConfigHandler.processConfigChanges来进行处理,方法里面再看看流程 ->
从动态配置entityConfig里面获取message.format.version配置消息格式版本号; 如果当前Broker的版本inter.broker.protocol.version 小于message.format.version配置；则将message.format.version配置排除掉
调用TopicConfigHandler.updateLogConfig 来更新指定Topic的所有TopicPartition的配置,其实是将TP正在加载或初始化的状态标记为没有完成初始化,这将会在后续过程中促成TP重新加载并初始化
将动态配置和并覆盖Server的默认配置为新的 newConfig, 然后根据Topic获取对应的Logs对象; 遍历Logs去更新newConfig；并尝试执行 initializeLeaderEpochCache； (需要注意的是：这里的动态配置不是支持所有的配置参数,请看【kafka运维】Kafka全网最全最详细运维命令合集(精品强烈建议收藏!!!)的附件部分)
当然特殊配置如leader.replication.throttled.replicas,follower.replication.throttled.replicas这两个限流相关；解析配置之后,然后通过quotaManager.markThrottled/quotaManager.removeThrottle更新/移除对应的限流分区集合
如果动态配置了unclean.leader.election.enable=true(允许非同步副本选主 )；那么就会执行TopicUncleanLeaderElectionEnable方法来让它改变选举策略(前提是当前Broker是Controller角色)

加载Broker动态配置

BrokerConfigHandler.processConfigChanges

假设我们配置了默认配置; zk里面的节点是<default>

sh bin/kafka-configs.sh –bootstrap-server xxxxx:9090 –alter –entity-type brokers --entity-default –add-config log.segment.bytes=88888888

从zk节点/config/changes 里面获取变更节点的json数据.然后去对应的 /config/{entityType}/{entituName}获取对应的数据
如果是<default>节点,说明有配置动态默认配置; 则按照静态配置<动态默认配置<动态指定配置的顺序重新加载覆盖一下; 如果新旧配置有变更(有可能执行了一次命令但是参数并没有变化的情况,修改了个寂寞)的情况下才会做更新的; 并且通知到所有的 BrokerReconfigurable; 这个就是上面启动时候 1.1 启动加载动态配置总流程的第2步骤 (注册可变更配置监听器) 注册的;
如果是指定BrokerId, 则除了上面2重新加载覆盖之外, 相关限流配置leader.replication.throttled.rate、follower.replication.throttled.rate、replica.alter.log.dirs.io.max.bytes.per.second 都会被更新一下quotaManagers.leader/leader/alterLogDirs.updateQuota ;如果这些配置没有配置的话,则用 Long.MaxValue(相当于是不限流)来更新

查询动态配置流程 `--describe`

简单检验
根据类型查询entities ; type是topics就获取所有topic; type是broker|broker-loggers则查询所有Broker节点
遍历entities获取配置；做些简单校验；然后想Broker发起describeConfigs请求; 节点策略是LeastLoadedNodeProvider
节点调用方法 KafkaApis.handleDescribeConfigsRequest
1. 未经授权配置不查询
2. 经过授权的配置开始查询 ;
3. 当查询的是topics时, 去zk节点/confgi/类型/类型名 ，获取到动态配置数据之后, 然后将其覆盖本地跟Log相关的静态配置，完事之后组装一下返回;(1.数据为空过滤2.敏感数据设置value=null; ConfigType=PASSWORD和不知道类型是啥的都是敏感数据 3. 组装所有的同义配置(静态默认配置、本地静态、默认动态配置、指定动态配置、等等多个配置))
  返回的数据类型如下:
如果有broker|broker-loggers节点, 则在获取到数据之后然后指定nodeId节点发起 describeBrokerConfigs请求
1. 如果查询的是brokers
2. 如果查询的是 broker-loggers

新增/修改/删除/动态配置的流程

1. 发起请求

查询当前的类型配置；这里的查询跟上面的--describe流程是一样的
相关校验；如果有delete-config配置, 需要校验一下当前配置有没有;如果没有抛出异常;
计算出需要变更的配置之后, 发起请求incrementalAlterConfigs；如果请求类型是 brokers/broker-loggers 则发起请求的接收方是指定的Broker 节点；否则就是LeastLoadedNodeProvider (当前负载最少的节点)

2. incrementalAlterConfigs 增量修改配置

KafkaApis.handleIncrementalAlterConfigsRequest

通过请求参数解析配置 configs
过滤一下未授权的配置
如果配置中有重复的项则抛出异常

Topic配置

获取节点 /config/topics/{topicName} 中的配置数据；
然后根据请求参数的属性，组装好变更后的配置是什么样的 configs；
简单校验一下, 并且支持自定义校验，如果有 alter.config.policy.class.name= 配置(默认null)的话,则会实例化指定的类(需要继承 AlterConfigPolicy类)；并调用他的 validate方法来校验;
调用写入zk配置的接口, 将动态配置重新写入(SetDataRequest)到接口 /config/topics/{topicName}中;
创建并写入配置变更记录顺序节点 /config/changes/config_change_序列号 中; 这个节点主要是让Broker们来监听这个节点的来了解到哪个配置有变更的;