TopicDeletionManager分析

TopicDeletionManager主要用于管理删除topic的状态机。

一核心字段

controllerContext: ControllerContext 用于维护KafkaController中上下文信息

partitionStateMachine: PartitionStateMachinepartition状态机

replicaStateMachine： ReplicaStateMahcine replica 状态机

topicsToBeDeleted：Set[String] 记录将要被删除topic集合

partitionsToBeDeleted:Set[TopicAndPartition] 记录将要被删除partition集合

topicsIneligibleForDeletion:Set[String] 用于记录不可删除的集合

deleteTopicStateChanged：AtomicBoolean 用于标记topic删除操作是否开始

deleteTopicsThread：DeleteTopicsThread 用于删除topic的后台线程

isDeleteTopicEnabled：Boolean 是否允许删除topic

二重要方法

2.1 start方法

defstart() {
// 判断是否支持删除topic
if (isDeleteTopicEnabled) {
    // 创建DeleteTopicsThread线程对象
    deleteTopicsThread= new DeleteTopicsThread()
    // 如果已经存在将要被删除的topic
    if (topicsToBeDeleted.nonEmpty)
      // 标记删除操作开始
      deleteTopicStateChanged.set(true)
    // 启动线程
    deleteTopicsThread.start()
}
}

2.2 DeleteTopicsThread的doWork方法使我们进行删除操作的入口方法

第一：获取待删除的topic的分区集合，发送UpdateMetadataCache

Request发送给所有的broker

private def onTopicDeletion(topics: Set[String]) {info("Topic deletion callback for %s".format(topics.mkString(",")))// 获取topic分区val partitions = topics.flatMap(controllerContext.partitionsForTopic)// 发送UpdateMetadataRequest请求controller.sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, partitions)// 获取所有AR并且按照topic分区val partitionReplicaAssignmentByTopic = controllerContext.partitionReplicaAssignment.groupBy(p => p._1.topic)topics.foreach { topic =>// 传入指定topic的AR副本集，开始分区的删除操作onPartitionDeletion(partitionReplicaAssignmentByTopic(topic).keySet)}
}

第二：调用onPartitionDeletion方法开始对指定分区进行删除

将不可用副本转化为ReplicaDeletionIneligible状态；

将可用副本转化为OfflineReplica状态，而且还会发送StopReplica

Request请求到带删除的副本，同时还会向broker发送跟新元数据的请求，将副本从ISR列表移除；

将可用副本由OfflineReplica转化为ReplicationDeletionStarted。

private def onPartitionDeletion(partitionsToBeDeleted: Set[TopicAndPartition]) {info("Partition deletion callback for %s".format(partitionsToBeDeleted.mkString(",")))// 获取每一个分区的副本val replicasPerPartition = controllerContext.replicasForPartition(partitionsToBeDeleted)// 开始删除startReplicaDeletion(replicasPerPartition)
}

private def startReplicaDeletion(replicasForTopicsToBeDeleted: Set[PartitionAndReplica]) {// 将要被删除的topic对应的分区副本按照topic分组replicasForTopicsToBeDeleted.groupBy(_.topic).foreach { case(topic, replicas) =>// 获取topic中所有可用的副本var aliveReplicasForTopic = controllerContext.allLiveReplicas().filter(p => p.topic.equals(topic))// 获取不可用的副本val deadReplicasForTopic = replicasForTopicsToBeDeleted -- aliveReplicasForTopicval successfullyDeletedReplicas = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionSuccessful)// 第一次删除的副本集合val replicasForDeletionRetry = aliveReplicasForTopic -- successfullyDeletedReplicas// 将不可用副本转换为ReplicaDeletionIneligible状态replicaStateMachine.handleStateChanges(deadReplicasForTopic, ReplicaDeletionIneligible)// 将待删除副本转化为OfflineReplicareplicaStateMachine.handleStateChanges(replicasForDeletionRetry, OfflineReplica)debug("Deletion started for replicas %s".format(replicasForDeletionRetry.mkString(",")))// 将待删除副本转化为ReplicaDeletionStartedcontroller.replicaStateMachine.handleStateChanges(replicasForDeletionRetry, ReplicaDeletionStarted,new Callbacks.CallbackBuilder().stopReplicaCallback(deleteTopicStopReplicaCallback).build)if(deadReplicasForTopic.nonEmpty) {debug("Dead Replicas (%s) found for topic %s".format(deadReplicasForTopic.mkString(","), topic))// 如果有副本挂掉，标记不能删除markTopicIneligibleForDeletion(Set(topic))}}
}

第三：调用StopReplicaResponse错误吗正常，则将副本转化为ReplicaDeletionSuccessful状态，否则转化为ReplicaDeletionIneligible状态，并标记该副本所在topic不能被删除，将该topic添加到topics

IneligibleForDeletion队列，最后唤醒DeleteTopicsThread

private def deleteTopicStopReplicaCallback(stopReplicaResponseObj: AbstractRequestResponse, replicaId: Int) {// 获取StopReplicaResponse响应val stopReplicaResponse = stopReplicaResponseObj.asInstanceOf[StopReplicaResponse]debug("Delete topic callback invoked for %s".format(stopReplicaResponse))val responseMap = stopReplicaResponse.responses.asScalaval partitionsInError =if (stopReplicaResponse.errorCode != Errors.NONE.code) responseMap.keySetelse responseMap.filter { case (_, error) => error != Errors.NONE.code }.keySetval replicasInError = partitionsInError.map(p => PartitionAndReplica(p.topic, p.partition, replicaId))inLock(controllerContext.controllerLock) {// move all the failed replicas to ReplicaDeletionIneligiblefailReplicaDeletion(replicasInError)// 如果响应没有异常if (replicasInError.size != responseMap.size) {val deletedReplicas = responseMap.keySet -- partitionsInErrorcompleteReplicaDeletion(deletedReplicas.map(p => PartitionAndReplica(p.topic, p.partition, replicaId)))}}
}

def failReplicaDeletion(replicas: Set[PartitionAndReplica]) {if(isDeleteTopicEnabled) {val replicasThatFailedToDelete = replicas.filter(r => isTopicQueuedUpForDeletion(r.topic))if(replicasThatFailedToDelete.nonEmpty) {val topics = replicasThatFailedToDelete.map(_.topic)debug("Deletion failed for replicas %s. Halting deletion for topics %s".format(replicasThatFailedToDelete.mkString(","), topics))controller.replicaStateMachine.handleStateChanges(replicasThatFailedToDelete, ReplicaDeletionIneligible)// 将当前线程标记为不可用markTopicIneligibleForDeletion(topics)// 唤醒DeleteTopicsThreadresumeTopicDeletionThread()}}
}

private def completeReplicaDeletion(replicas: Set[PartitionAndReplica]) {val successfullyDeletedReplicas = replicas.filter(r => isTopicQueuedUpForDeletion(r.topic))debug("Deletion successfully completed for replicas %s".format(successfullyDeletedReplicas.mkString(",")))// 成功的将删除的副本转化为ReplicaDeletionSuccessful状态controller.replicaStateMachine.handleStateChanges(successfullyDeletedReplicas, ReplicaDeletionSuccessful)resumeTopicDeletionThread() // 唤醒DeleteTopicThread
}

第四：开始第二次调用doWork方法，如果副本已经处于Replica

DeletionSuccessful状态，调用completeDeleteTopic方法完成topic删除

private def completeDeleteTopic(topic: String) {// 取消partitionModificationListener的监听partitionStateMachine.deregisterPartitionChangeListener(topic)// 获取已经删除topic的副本val replicasForDeletedTopic = controller.replicaStateMachine.replicasInState(topic, ReplicaDeletionSuccessful)// 将状态转化为NonExistentReplica对象replicaStateMachine.handleStateChanges(replicasForDeletedTopic, NonExistentReplica)val partitionsForDeletedTopic = controllerContext.partitionsForTopic(topic)// 将topic对应的parition 转化为offlinePartitionpartitionStateMachine.handleStateChanges(partitionsForDeletedTopic, OfflinePartition)// 紧接着转化为NonExistentPartitionpartitionStateMachine.handleStateChanges(partitionsForDeletedTopic, NonExistentPartition)// 将该topic从待删除队里移除topicsToBeDeleted -= topicpartitionsToBeDeleted.retain(_.topic != topic)val zkUtils = controllerContext.zkUtilszkUtils.zkClient.deleteRecursive(getTopicPath(topic))zkUtils.zkClient.deleteRecursive(getEntityConfigPath(ConfigType.Topic, topic))zkUtils.zkClient.delete(getDeleteTopicPath(topic))// 从ControllerContext中移除该topiccontrollerContext.removeTopic(topic)
}

TopicDeletionManager分析相关推荐

Kafka源码分析10：副本状态机ReplicaStateMachine详解（图解+秒懂+史上最全）
文章很长,建议收藏起来,慢慢读! Java 高并发发烧友社群:疯狂创客圈奉上以下珍贵的学习资源: 免费赠送经典图书:<Java高并发核心编程(卷1)> 面试必备 + 大厂必备 +涨薪 ...
Kafka分区副本重分配源码分析
Kafka分区副本重分配文章目录 Kafka分区副本重分配 1.前言 2.分区副本重分配流程图 3.分区副本重分配详细分析 3.1 客户端行为 3.1.1 执行副本重分配脚本 3.1.2 解析并验证 ...
【Golang源码分析】Go Web常用程序包gorilla/mux的使用与源码简析
目录[阅读时间:约10分钟] 一.概述二.对比: gorilla/mux与net/http DefaultServeMux 三.简单使用四.源码简析 1.NewRouter函数 2.HandleF ...
2022-2028年中国自动驾驶系统行业现状调研分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了中国自动驾驶系统行业市场行业相关概述.中国自 ...
2022-2028年中国阻尼涂料市场研究及前瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了中国阻尼涂料行业市场行业相关概述.中国阻尼涂 ...
2021-2028年中国阻燃装饰行业市场需求与投资规划分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了中国阻燃装饰行业市场行业相关概述.中国阻燃装 ...
2022-2028年全球与中国漂白吸水棉市场研究及前瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了全球与中国漂白吸水棉行业市场行业相关概述.全 ...
2022-2028年全球与中国青苔清洗剂市场研究及前瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了全球与中国青苔清洗剂行业市场行业相关概述.全 ...
2022-2028年全球与中国氢碘化物市场智研瞻分析报告
[报告类型]产业研究 [报告价格]4500起 [出版时间]即时更新(交付时间约3个工作日) [发布机构]智研瞻产业研究院 [报告格式]PDF版本报告介绍了全球与中国氢碘化物行业市场行业相关概述.全球 ...

TopicDeletionManager分析

TopicDeletionManager分析相关推荐

最新文章

热门文章