上篇讲述了 KafkaController 的启动流程,但是关于分区状态机和副本状态机的初始化并没有触及,分区状态机和副本状态机的内容将在本篇文章深入讲述。分区状态机记录着当前集群所有 Partition 的状态信息以及如何对 Partition 状态转移进行相应的处理;副本状态机则是记录着当前集群所有 Replica 的状态信息以及如何对 Replica 状态转变进行相应的处理。

ReplicaStateMachine

ReplicaStateMachine 记录着集群所有 Replica 的状态信息,它决定着一个 replica 处在什么状态以及它在什么状态下可以转变为什么状态,Kafka 中副本的状态总共有以下七种类型:

  1. NewReplica:这种状态下 Controller 可以创建这个 Replica,这种状态下该 Replica 只能作为 follower,它可以是 Replica 删除后的一个临时状态,它有效的前置状态是 NonExistentReplica;
  2. OnlineReplica:一旦这个 Replica 被分配到指定的 Partition 上,并且 Replica 创建完成,那么它将会被置为这个状态,在这个状态下,这个 Replica 既可以作为 leader 也可以作为 follower,它有效的前置状态是 NewReplica、OnlineReplica 或 OfflineReplica;
  3. OfflineReplica:如果一个 Replica 挂掉(所在的节点宕机或者其他情况),该 Replica 将会被转换到这个状态,它有的效前置状态是 NewReplica、OfflineReplica 或者 OnlineReplica;
  4. ReplicaDeletionStarted:Replica 开始删除时被置为的状态,它有效的前置状态是 OfflineReplica;
  5. ReplicaDeletionSuccessful:如果 Replica 在删除时没有遇到任何错误信息,它将被置为这个状态,这个状态代表该 Replica 的数据已经从节点上清除了,它有效的前置状态是 ReplicaDeletionStarted;
  6. ReplicaDeletionIneligible:如果 Replica 删除失败,它将会转移到这个状态,这个状态意思是非法删除,也就是删除是无法成功的,它有效的前置状态是 ReplicaDeletionStarted;
  7. NonExistentReplica:如果 Replica 删除成功,它将被转移到这个状态,它有效的前置状态是:ReplicaDeletionSuccessful。

上面的状态中其中后面4是专门为 Replica 删除而服务的,副本状态机转移图如下所示:

副本状态机

这张图是副本状态机的核心,在下面会详细讲述,接下来先看下 KafkaController 在启动时,调用 ReplicaStateMachine 的 startup() 方法初始化的处理过程。

ReplicaStateMachine 初始化

副本状态机初始化的过程如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
//note: Controller 重新选举后触发的操作
def startup() {
// initialize replica state
//note: 初始化 zk 上所有的 Replica 状态信息(replica 存活的话设置为 Online,不存活的设置为 ReplicaDeletionIneligible)
initializeReplicaState()
// set started flag
hasStarted.set(true)
// move all Online replicas to Online
//note: 将存活的副本状态转变为 OnlineReplica
handleStateChanges(controllerContext.allLiveReplicas(), OnlineReplica)
info("Started replica state machine with initial state -> " + replicaState.toString())
}

在这个方法中,ReplicaStateMachine 先调用 initializeReplicaState() 方法初始化集群中所有 Replica 的状态信息,如果 Replica 所在机器是 alive 的,那么将其状态设置为 OnlineReplica,否则设置为 ReplicaDeletionIneligible 状态,这里只是将 Replica 的状态信息更新副本状态机的缓存 replicaState 中,并没有真正进行状态转移的操作。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
//note: 初始化所有副本的状态信息
private def initializeReplicaState() {
for((topicPartition, assignedReplicas) <- controllerContext.partitionReplicaAssignment) {
val topic = topicPartition.topic
val partition = topicPartition.partition
assignedReplicas.foreach { replicaId =>
val partitionAndReplica = PartitionAndReplica(topic, partition, replicaId)
if (controllerContext.liveBrokerIds.contains(replicaId)) //note: 如果副本是存活,那么将状态都设置为 OnlineReplica
replicaState.put(partitionAndReplica, OnlineReplica)
else
// mark replicas on dead brokers as failed for topic deletion, if they belong to a topic to be deleted.
// This is required during controller failover since during controller failover a broker can go down,
// so the replicas on that broker should be moved to ReplicaDeletionIneligible to be on the safer side.
//note: 将不存活的副本状态设置为 ReplicaDeletionIneligible
replicaState.put(partitionAndReplica, ReplicaDeletionIneligible)
}
}
}

接着第二步调用 handleStateChanges() 将所有存活的副本状态转移为 OnlineReplica 状态,这里才是真正进行状态转移的地方,其具体实现如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
//note: 用于处理 Replica 状态的变化
def handleStateChanges(replicas: Set[PartitionAndReplica], targetState: ReplicaState,
callbacks: Callbacks = (new CallbackBuilder).build) {
if(replicas.nonEmpty) {
info("Invoking state change to %s for replicas %s".format(targetState, replicas.mkString(",")))
try {
brokerRequestBatch.newBatch()
//note: 状态转变
replicas.foreach(r => handleStateChange(r, targetState, callbacks))
//note: 向 broker 发送相应请求
brokerRequestBatch.sendRequestsToBrokers(controller.epoch)
}catch {
case e: Throwable => error("Error while moving some replicas to %s state".format(targetState), e)
}
}
}

这里是副本状态机 startup() 方法的最后一步,它的目的是将所有 alive 的 Replica 状态转移到 OnlineReplica 状态,由于前面已经这些 alive replica 的状态设置成了 OnlineReplica,所以这里 Replica 的状态转移情况是:OnlineReplica –> OnlineReplica,这个方法主要是做了两件事:

  1. 状态转移(这个在下面详细讲述);
  2. 发送相应的请求。

副本的状态转移

这里以要转移的 TargetState 区分做详细详细讲解,当 TargetState 分别是 NewReplica、ReplicaDeletionStarted、ReplicaDeletionIneligible、ReplicaDeletionSuccessful、NonExistentReplica、OnlineReplica 或者 OfflineReplica 时,副本状态机所做的事情。

TargetState: NewReplica

NewReplica 这个状态是 Replica 准备开始创建是的一个状态,其实现逻辑如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
val currState = replicaState.getOrElseUpdate(partitionAndReplica, NonExistentReplica)//note: Replica 不存在的话,状态初始化为 NonExistentReplica
assertValidPreviousStates(partitionAndReplica, List(NonExistentReplica), targetState)//note: 验证
// start replica as a follower to the current leader for its partition
//note: 从 zk 获取 Partition 的 leaderAndIsr 信息
val leaderIsrAndControllerEpochOpt = ReplicationUtils.getLeaderIsrAndEpochForPartition(zkUtils, topic, partition)
leaderIsrAndControllerEpochOpt match {
case Some(leaderIsrAndControllerEpoch) =>
if(leaderIsrAndControllerEpoch.leaderAndIsr.leader == replicaId)//note: 这个状态的 Replica 不能作为 leader
throw new StateChangeFailedException("Replica %d for partition %s cannot be moved to NewReplica"
.format(replicaId, topicAndPartition) + "state as it is being requested to become leader")
//note: 向该 replicaId 发送 LeaderAndIsr 请求,这个方法同时也会向所有的 broker 发送 updateMeta 请求
brokerRequestBatch.addLeaderAndIsrRequestForBrokers(List(replicaId),
topic, partition, leaderIsrAndControllerEpoch,
replicaAssignment)
case None => // new leader request will be sent to this replica when one gets elected
}
replicaState.put(partitionAndReplica, NewReplica)//note: 缓存这个 replica 对象的状态
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState,
targetState))

当想要把 Replica 的状态转移为 NewReplica 时,副本状态机的处理逻辑如下:

  1. 校验 Replica 的前置状态,只有处于 NonExistentReplica 状态的副本才能转移到 NewReplica 状态;
  2. 从 zk 中获取该 Topic-Partition 的 LeaderIsrAndControllerEpoch 信息;
  3. 如果获取不到上述信息,直接将该 Replica 的状态转移成 NewReplica,然后结束流程(对与新建的 Partition,处于这个状态时,该 Partition 是没有相应的 LeaderAndIsr 信息的);
  4. 获取到 Partition 的 LeaderIsrAndControllerEpoch 信息,如果发现该 Partition 的 leader 是当前副本,那么就抛出 StateChangeFailedException 异常,因为处在这个状态的 Replica 是不能被选举为 leader 的;
  5. 获取到了 Partition 的 LeaderIsrAndControllerEpoch 信息,并且该 Partition 的 leader 不是当前 replica,那么向该 Partition 的所有 Replica 添加一个 LeaderAndIsr 请求(添加 LeaderAndIsr 请求时,实际上也会向所有的 Broker 都添加一个 Update-Metadata 请求);
  6. 最后将该 Replica 的状态转移成 NewReplica,然后结束流程。

TargetState: ReplicaDeletionStarted

这是 Replica 开始删除时的状态,Replica 转移到这种状态的处理实现如下:

1
2
3
4
5
6
7
8
assertValidPreviousStates(partitionAndReplica, List(OfflineReplica), targetState)
replicaState.put(partitionAndReplica, ReplicaDeletionStarted)
// send stop replica command
//note: 发送 StopReplica 请求给该副本,并设置 deletePartition=true
brokerRequestBatch.addStopReplicaRequestForBrokers(List(replicaId), topic, partition, deletePartition = true,
callbacks.stopReplicaResponseCallback)
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState, targetState))

这部分的实现逻辑:

  1. 校验其前置状态,Replica 只能是在 OfflineReplica 的情况下才能转移到这种状态;
  2. 更新向该 Replica 的状态为 ReplicaDeletionStarted;
  3. 向该 replica 发送 StopReplica 请求(deletePartition = true),收到这请求后,broker 会从物理存储上删除这个 Replica 的数据内容;
  4. 如果请求返回的话会触发其回调函数(这部分会在 topic 删除部分讲解)。

TargetState: ReplicaDeletionIneligible

ReplicaDeletionIneligible 是副本删除失败时的状态,Replica 转移到这种状态的处理实现如下:

1
2
3
4
assertValidPreviousStates(partitionAndReplica, List(ReplicaDeletionStarted), targetState)
replicaState.put(partitionAndReplica, ReplicaDeletionIneligible)
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState, targetState))

实现逻辑:

  1. 校验其前置状态,Replica 只能是在 ReplicaDeletionStarted 下才能转移这种状态;
  2. 更新该 Replica 的状态为 ReplicaDeletionIneligible。

TargetState: ReplicaDeletionSuccessful

ReplicaDeletionSuccessful 是副本删除成功时的状态,Replica 转移到这种状态的处理实现如下:

1
2
3
4
assertValidPreviousStates(partitionAndReplica, List(ReplicaDeletionStarted), targetState)
replicaState.put(partitionAndReplica, ReplicaDeletionSuccessful)
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState, targetState))

实现逻辑:

  1. 检验其前置状态,Replica 只能是在 ReplicaDeletionStarted 下才能转移这种状态;
  2. 更新该 Replica 的状态为 ReplicaDeletionSuccessful。

TargetState: NonExistentReplica

NonExistentReplica 是副本完全删除、不存在这个副本的状态,Replica 转移到这种状态的处理实现如下:

1
2
3
4
5
6
7
8
assertValidPreviousStates(partitionAndReplica, List(ReplicaDeletionSuccessful), targetState)
// remove this replica from the assigned replicas list for its partition
val currentAssignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
//note: 从 controller 和副本状态机的缓存中清除这个 Replica 的记录西溪
controllerContext.partitionReplicaAssignment.put(topicAndPartition, currentAssignedReplicas.filterNot(_ == replicaId))
replicaState.remove(partitionAndReplica)
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState, targetState))

实现逻辑:

  1. 检验其前置状态,Replica 只能是在 ReplicaDeletionSuccessful 下才能转移这种状态;
  2. 在 controller 的 partitionReplicaAssignment 删除这个 Partition 对应的 replica 信息;
  3. 从 Controller 和副本状态机中将这个 Topic 从缓存中删除。

TargetState: OnlineReplica

OnlineReplica 是副本正常工作时的状态,此时的 Replica 既可以作为 leader 也可以作为 follower,Replica 转移到这种状态的处理实现如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
assertValidPreviousStates(partitionAndReplica,
List(NewReplica, OnlineReplica, OfflineReplica, ReplicaDeletionIneligible), targetState)
replicaState(partitionAndReplica) match {
case NewReplica => //note: NewReplica --> OnlineReplica
// add this replica to the assigned replicas list for its partition
//note: 向 the assigned replicas list 添加这个 replica(正常情况下这些 replicas 已经更新到 list 中了)
val currentAssignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
if(!currentAssignedReplicas.contains(replicaId))
controllerContext.partitionReplicaAssignment.put(topicAndPartition, currentAssignedReplicas :+ replicaId)
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState,
targetState))
case _ => //note: OnlineReplica/OfflineReplica/ReplicaDeletionIneligible --> OnlineReplica
// check if the leader for this partition ever existed
//note: 如果该 Partition 的 LeaderIsrAndControllerEpoch 信息存在,那么就更新副本的状态,并发送相应的请求
controllerContext.partitionLeadershipInfo.get(topicAndPartition) match {
case Some(leaderIsrAndControllerEpoch) =>
brokerRequestBatch.addLeaderAndIsrRequestForBrokers(List(replicaId), topic, partition, leaderIsrAndControllerEpoch,
replicaAssignment)
replicaState.put(partitionAndReplica, OnlineReplica)
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState, targetState))
case None => // that means the partition was never in OnlinePartition state, this means the broker never
// started a log for that partition and does not have a high watermark value for this partition
}
}
replicaState.put(partitionAndReplica, OnlineReplica)

从前面的状态转移图中可以看出,当 Replica 处在 NewReplica、OnlineReplica、OfflineReplica 或者 ReplicaDeletionIneligible 状态时,Replica 是可以转移到 OnlineReplica 状态的,下面分两种情况讲述:

NewReplica –> OnlineReplica 的处理逻辑如下:

  1. 从 Controller 的 partitionReplicaAssignment 中获取这个 Partition 的 AR;
  2. 如果 Replica 不在 AR 中的话,那么就将其添加到 Partition 的 AR 中;
  3. 最后将 Replica 的状态设置为 OnlineReplica 状态。

OnlineReplica/OfflineReplica/ReplicaDeletionIneligible –> OnlineReplica 的处理逻辑如下:

  1. 从 Controller 的 partitionLeadershipInfo 中获取 Partition 的 LeaderAndIsr 信息;
  2. 如果该信息存在,那么就向这个 Replica 所在 broker 添加这个 Partition 的 LeaderAndIsr 请求,并将 Replica 的状态设置为 OnlineReplica 状态;
  3. 否则不做任务处理;
  4. 最后更新R Replica 的状态为 OnlineReplica。

TargetState: OfflineReplica

OfflineReplica 是 Replica 所在 Broker 掉线时 Replica 的状态,转移到这种状态的处理逻辑如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
assertValidPreviousStates(partitionAndReplica,
List(NewReplica, OnlineReplica, OfflineReplica, ReplicaDeletionIneligible), targetState)
// send stop replica command to the replica so that it stops fetching from the leader
//note: 发送 StopReplica 请求给该副本,先停止副本同步
brokerRequestBatch.addStopReplicaRequestForBrokers(List(replicaId), topic, partition, deletePartition = false)
// As an optimization, the controller removes dead replicas from the ISR
val leaderAndIsrIsEmpty: Boolean =
controllerContext.partitionLeadershipInfo.get(topicAndPartition) match {
case Some(_) =>
controller.removeReplicaFromIsr(topic, partition, replicaId) match { //note: 从 isr 中移除这个副本(前提是 ISR 有其他有效副本)
case Some(updatedLeaderIsrAndControllerEpoch) =>
// send the shrunk ISR state change request to all the remaining alive replicas of the partition.
//note: 发送 LeaderAndIsr 请求给剩余的其他副本,因为 ISR 变动了
val currentAssignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition)
if (!controller.deleteTopicManager.isPartitionToBeDeleted(topicAndPartition)) {
brokerRequestBatch.addLeaderAndIsrRequestForBrokers(currentAssignedReplicas.filterNot(_ == replicaId),
topic, partition, updatedLeaderIsrAndControllerEpoch, replicaAssignment)
}
replicaState.put(partitionAndReplica, OfflineReplica)
stateChangeLogger.trace("Controller %d epoch %d changed state of replica %d for partition %s from %s to %s"
.format(controllerId, controller.epoch, replicaId, topicAndPartition, currState, targetState))
false
case None =>
true
}
case None =>
true
}
if (leaderAndIsrIsEmpty && !controller.deleteTopicManager.isPartitionToBeDeleted(topicAndPartition))
throw new StateChangeFailedException(
"Failed to change state of replica %d for partition %s since the leader and isr path in zookeeper is empty"
.format(replicaId, topicAndPartition))

处理逻辑如下:

  1. 校验其前置状态,只有 Replica 在 NewReplica、OnlineReplica、OfflineReplica 或者 ReplicaDeletionIneligible 状态时,才能转移到这种状态;
  2. 向该 Replica 所在节点发送 StopReplica 请求(deletePartition = false);
  3. 调用 Controller 的 removeReplicaFromIsr() 方法将该 replica 从 Partition 的 isr 移除这个 replica(前提 isr 中还有其他有效副本),然后向该 Partition 的其他副本发送 LeaderAndIsr 请求;
  4. 更新这个 Replica 的状态为 OfflineReplica。

状态转移触发的条件

这里主要是看一下上面 Replica 各种转移的触发的条件,整理的结果如下表所示,部分内容会在后续文章讲解。

TargetState 触发方法 作用
OnlineReplica KafkaController 的 onBrokerStartup() Broker 启动时,目的是将在该节点的 Replica 状态设置为 OnlineReplica
OnlineReplica KafkaController 的 onNewPartitionCreation() 新建 Partition 时,Replica 初始化及 Partition 状态变成 OnlinePartition 后,新创建的 Replica 状态也变为 OnlineReplica;
OnlineReplica KafkaController 的 onPartitionReassignment() 副本迁移完成后,RAR 中的副本设置为 OnlineReplica 状态
OnlineReplica ReplicaStateMachine 的 startup() 副本状态机刚初始化启动时,将存活的副本状态设置为 OnlineReplica
OfflineReplica TopicDeletionManager 的 markTopicForDeletionRetry() 将删除失败的 Replica 设置为 OfflineReplica,重新进行删除
OfflineReplica TopicDeletionManager 的 startReplicaDeletion() 开始副本删除时,先将副本设置为 OfflineReplica
OfflineReplica KafkaController 的 shutdownBroker() 方法 优雅关闭 broker 时,目的是把下线节点上的副本状态设置为 OfflineReplica
OfflineReplica KafkaController 的 onBrokerFailure() broker 掉线时,目的是把下线节点上的副本状态设置为 OfflineReplica
NewReplica KafkaController 的 onNewPartitionCreation() Partition 新建时,当 Partition 状态变为 NewPartition 后,副本的状态变为 NewReplica
NewReplica KafkaController 的 startNewReplicasForReassignedPartition() Partition 副本迁移时,将新分配的副本状态设置为 NewReplica;
ReplicaDeletionStarted TopicDeletionManager 的 startReplicaDeletion() 下线副本时,将成功设置为 OfflineReplica 的 Replica 设置为 ReplicaDeletionStarted 状态,开始物理上删除副本数据(也是发送 StopReplica)
ReplicaDeletionStarted KafkaController 的 stopOldReplicasOfReassignedPartition() Partition 的副本迁移时,目的是下线那些 old replica,新的 replica 已经迁移到新分配的副本上了
ReplicaDeletionSuccessful TopicDeletionManager 的 completeReplicaDeletion() 物理将数据成功删除的 Replica 状态会变为这个
ReplicaDeletionSuccessful KafkaController 的 stopOldReplicasOfReassignedPartition() Partition 的副本迁移时,在下线那些旧 Replica 时的一个状态,删除成功
ReplicaDeletionIneligible TopicDeletionManager 的 startReplicaDeletion() 开始副本删除时,删除失败的副本会设置成这个状态
ReplicaDeletionIneligible KafkaController 的 stopOldReplicasOfReassignedPartition() Partition 副本迁移时,在下线那些旧的 Replica 时的一个状态,删除失败
NonExistentReplica TopicDeletionManager 的 completeReplicaDeletion() 副本删除成功后(状态为 ReplicaDeletionSuccessful),从状态机和 Controller 的缓存中清除该副本的记录;
NonExistentReplica KafkaController 的 stopOldReplicasOfReassignedPartition() Partition 的副本成功迁移、旧副本成功删除后,从状态机和 Controller 的缓存中清除旧副本的记录

PartitionStateMachine

PartitionStateMachine 记录着集群所有 Partition 的状态信息,它决定着一个 Partition 处在什么状态以及它在什么状态下可以转变为什么状态,Kafka 中副本的状态总共有以下四种类型:

  1. NonExistentPartition:这个代表着这个 Partition 之前没有被创建过或者之前创建了现在又被删除了,它有效的前置状态是 OfflinePartition;
  2. NewPartition:Partition 创建后,它将处于这个状态,这个状态的 Partition 还没有 leader 和 isr,它有效的前置状态是 NonExistentPartition;
  3. OnlinePartition:一旦这个 Partition 的 leader 被选举出来了,它将处于这个状态,它有效的前置状态是 NewPartition、OnlinePartition、OfflinePartition;
  4. OfflinePartition:如果这个 Partition 的 leader 掉线,这个 Partition 将被转移到这个状态,它有效的前置状态是 NewPartition、OnlinePartition、OfflinePartition。

分区状态机转移图如下所示:

分区状态机

这张图是分区状态机的核心,在下面会详细讲述,接下来先看下 KafkaController 在启动时,调用 PartitionStateMachine 的 startup() 方法初始化的处理过程。

PartitionStateMachine 初始化

PartitionStateMachine 的初始化方法如下所示:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
//note: Controller 启动时触发
//note: 初始化所有 Partition 的状态(从 zk 获取), 然后对于 new/offline Partition 触发选主(选主成功的话,变为 OnlinePartition)
def startup() {
// initialize partition state
//note: 初始化 partition 的状态,如果 leader 所在 broker 是 alive 的,那么状态为 OnlinePartition,否则为 OfflinePartition
initializePartitionState()
// set started flag
hasStarted.set(true)
// try to move partitions to online state
//note: 为所有处理 NewPartition 或 OnlinePartition 状态 Partition 选举 leader
triggerOnlinePartitionStateChange()
info("Started partition state machine with initial state -> " + partitionState.toString())
}

在这个方法中,PartitionStateMachine 先调用 initializePartitionState() 方法初始化集群中所有 Partition 的状态信息:

  1. 如果该 Partition 有 LeaderAndIsr 信息,那么如果 Partition leader 所在的机器是 alive 的,那么将其状态设置为 OnlinePartition,否则设置为 OfflinePartition 状态;
  2. 如果该 Partition 没有 LeaderAndIsr 信息,那么将其状态设置为 NewPartition。

这里只是将 Partition 的状态信息更新分区状态机的缓存 partitionState 中,并没有真正进行状态的转移。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
//note: 根据从 zk 获取的所有 Partition,进行状态初始化
private def initializePartitionState() {
for (topicPartition <- controllerContext.partitionReplicaAssignment.keys) {
// check if leader and isr path exists for partition. If not, then it is in NEW state
controllerContext.partitionLeadershipInfo.get(topicPartition) match {
case Some(currentLeaderIsrAndEpoch) =>
// else, check if the leader for partition is alive. If yes, it is in Online state, else it is in Offline state
if (controllerContext.liveBrokerIds.contains(currentLeaderIsrAndEpoch.leaderAndIsr.leader))
// leader is alive
//note: 有 LeaderAndIsr 信息,并且 leader 存活,设置为 OnlinePartition 状态
partitionState.put(topicPartition, OnlinePartition)
else
//note: 有 LeaderAndIsr 信息,但是 leader 不存活,设置为 OfflinePartition 状态
partitionState.put(topicPartition, OfflinePartition)
case None =>
//note: 没有 LeaderAndIsr 信息,设置为 NewPartition 状态(这个 Partition 还没有)
partitionState.put(topicPartition, NewPartition)
}
}
}

在初始化的第二步,将会调用 triggerOnlinePartitionStateChange() 方法,为所有的状态为 NewPartition/OnlinePartition 的 Partition 进行 leader 选举,选举成功后的话,其状态将会设置为 OnlinePartition,调用的 Leader 选举方法是 OfflinePartitionLeaderSelector(具体实现参考链接)。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
//note: 这个方法是在 controller 选举后或 broker 上线或下线时时触发的
def triggerOnlinePartitionStateChange() {
try {
brokerRequestBatch.newBatch()
// try to move all partitions in NewPartition or OfflinePartition state to OnlinePartition state except partitions
// that belong to topics to be deleted
//note: 开始为所有状态在 NewPartition or OfflinePartition 状态的 partition 更新状态(除去将要被删除的 topic)
for((topicAndPartition, partitionState) <- partitionState
if !controller.deleteTopicManager.isTopicQueuedUpForDeletion(topicAndPartition.topic)) {
if(partitionState.equals(OfflinePartition) || partitionState.equals(NewPartition))
//note: 尝试为处在 OfflinePartition 或 NewPartition 状态的 Partition 选主,成功后转换为 OnlinePartition
handleStateChange(topicAndPartition.topic, topicAndPartition.partition, OnlinePartition, controller.offlinePartitionSelector,
(new CallbackBuilder).build)
}
//note: 发送请求给所有的 broker,包括 LeaderAndIsr 请求和 UpdateMetadata 请求(这里只是添加到 Broker 对应的 RequestQueue 中,后台有线程去发送)
brokerRequestBatch.sendRequestsToBrokers(controller.epoch)
} catch {
case e: Throwable => error("Error while moving some partitions to the online state", e)
// TODO: It is not enough to bail out and log an error, it is important to trigger leader election for those partitions
}
}

上面方法的目的是为尝试将所有的状态为 NewPartition/OnlinePartition 的 Partition 状态转移到 OnlinePartition,这个方法主要是做了两件事:

  1. 状态转移(这个在下面详细讲述);
  2. 发送相应的请求。

分区的状态转移

这里以要转移的 TargetState 区分做详细详细讲解,当 TargetState 分别是 NewPartition、OfflinePartition、NonExistentPartition 或者 OnlinePartition 时,副本状态机所做的事情。

TargetState: NewPartition

NewPartition 是 Partition 刚创建时的一个状态,其处理逻辑如下:

1
2
3
4
5
6
7
8
9
//note: 如果该 Partition 的状态不存在,默认为 NonExistentPartition
val currState = partitionState.getOrElseUpdate(topicAndPartition, NonExistentPartition)
// pre: partition did not exist before this
assertValidPreviousStates(topicAndPartition, List(NonExistentPartition), NewPartition)
partitionState.put(topicAndPartition, NewPartition) //note: 缓存 partition 的状态
val assignedReplicas = controllerContext.partitionReplicaAssignment(topicAndPartition).mkString(",")
stateChangeLogger.trace("Controller %d epoch %d changed partition %s state from %s to %s with assigned replicas %s"
.format(controllerId, controller.epoch, topicAndPartition, currState, targetState,
assignedReplicas))

实现逻辑:

  1. 校验其前置状态,它有效的前置状态为 NonExistentPartition;
  2. 将该 Partition 的状态转移为 NewPartition 状态,并且更新到缓存中。

TargetState: OnlinePartition

OnlinePartition 是一个 Partition 正常工作时的状态,这个状态下的 Partition 已经成功选举出了 leader 和 isr 信息,其实现逻辑如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
//note: 判断 Partition 之前的状态是否可以转换为目的状态
assertValidPreviousStates(topicAndPartition, List(NewPartition, OnlinePartition, OfflinePartition), OnlinePartition)
partitionState(topicAndPartition) match {
case NewPartition => //note: 新建的 Partition
//note: 选举 leader 和 isr,更新到 zk 和 controller 中,如果没有存活的 replica,抛出异常
// initialize leader and isr path for new partition
initializeLeaderAndIsrForPartition(topicAndPartition)
case OfflinePartition => //note: leader 挂掉的 Partition
//note: 进行 leader 选举,更新到 zk 及 controller 缓存中,失败的抛出异常
electLeaderForPartition(topic, partition, leaderSelector)
case OnlinePartition => // invoked when the leader needs to be re-elected
//note:这种只有在 leader 需要重新选举时才会触发
electLeaderForPartition(topic, partition, leaderSelector)
case _ => // should never come here since illegal previous states are checked above
}
partitionState.put(topicAndPartition, OnlinePartition)
val leader = controllerContext.partitionLeadershipInfo(topicAndPartition).leaderAndIsr.leader
stateChangeLogger.trace("Controller %d epoch %d changed partition %s from %s to %s with leader %d"
.format(controllerId, controller.epoch, topicAndPartition, currState, targetState, leader))

实现逻辑:

  1. 校验这个 Partition 的前置状态,有效的前置状态是:NewPartition、OnlinePartition 或者 OfflinePartition;
  2. 如果前置状态是 NewPartition,那么为该 Partition 选举 leader 和 isr,更新到 zk 和 controller 的缓存中,如果副本没有处于 alive 状态的话,就抛出异常;
  3. 如果前置状态是 OnlinePartition,那么只是触发 leader 选举,在 OnlinePartition –> OnlinePartition 这种状态转移时,需要传入 leader 选举的方法,触发该 Partition 的 leader 选举;
  4. 如果前置状态是 OfflinePartition,同上,也是触发 leader 选举。
  5. 更新 Partition 的状态为 OnlinePartition。

对于以上这几种情况,无论前置状态是什么,最后都会触发这个 Partition 的 leader 选举,leader 成功后,都会触发向这个 Partition 的所有 replica 发送 LeaderAndIsr 请求。

TargetState: OfflinePartition

OfflinePartition 是这个 Partition 的 leader 挂掉时转移的一个状态,如果 Partition 转移到这个状态,那么就意味着这个 Partition 没有了可用 leader。

1
2
3
4
5
6
7
// pre: partition should be in New or Online state
assertValidPreviousStates(topicAndPartition, List(NewPartition, OnlinePartition, OfflinePartition), OfflinePartition)
// should be called when the leader for a partition is no longer alive
stateChangeLogger.trace("Controller %d epoch %d changed partition %s state from %s to %s"
.format(controllerId, controller.epoch, topicAndPartition, currState, targetState))
partitionState.put(topicAndPartition, OfflinePartition)
// post: partition has no alive leader

实现逻辑:

  1. 校验其前置状态,它有效的前置状态为 NewPartition、OnlinePartition 或者 OfflinePartition;
  2. 将该 Partition 的状态转移为 OfflinePartition 状态,并且更新到缓存中。

TargetState: NonExistentPartition

NonExistentPartition 代表了已经处于 OfflinePartition 状态的 Partition 已经从 metadata 和 zk 中删除后进入的状态。

1
2
3
4
5
6
// pre: partition should be in Offline state
assertValidPreviousStates(topicAndPartition, List(OfflinePartition), NonExistentPartition)
stateChangeLogger.trace("Controller %d epoch %d changed partition %s state from %s to %s"
.format(controllerId, controller.epoch, topicAndPartition, currState, targetState))
partitionState.put(topicAndPartition, NonExistentPartition)
// post: partition state is deleted from all brokers and zookeeper

实现逻辑:

  1. 校验其前置状态,它有效的前置状态为 OfflinePartition;
  2. 将该 Partition 的状态转移为 NonExistentPartition 状态,并且更新到缓存中。

状态转移触发的条件

这里主要是看一下上面 Partition 各种转移的触发的条件,整理的结果如下表所示,部分内容会在后续文章讲解。

TargetState 触发方法 作用
OnlinePartition Controller 的 shutdownBroker() 优雅关闭 Broker 时调用,因为要下线的节点是 leader,所以需要触发 leader 选举
OnlinePartition Controller 的 onNewPartitionCreation() Partition 新建时,这个是在 Replica 已经变为 NewPartition 状态后进行的,为新建的 Partition 初始化 leader 和 isr
OnlinePartition controller 的 onPreferredReplicaElection() 对 Partition 进行最优 leader 选举,目的是触发 leader 选举
OnlinePartition controller 的 moveReassignedPartitionLeaderIfRequired() 分区副本迁移完成后,1. 当前的 leader 不在 RAR 中,需要触发 leader 选举;2. 当前 leader 在 RAR 但是掉线了,也需要触发 leader 选举
OnlinePartition PartitionStateMachine 的 triggerOnlinePartitionStateChange() 当 Controller 重新选举出来或 broker 有变化时,目的为了那些状态为 NewPartition/OfflinePartition 的 Partition 重新选举 leader,选举成功后状态变为 OnlinePartition
OnlinePartition PartitionStateMachine 的 initializePartitionState() Controller 初始化时,遍历 zk 的所有的分区,如果有 LeaderAndIsr 信息并且 leader 在 alive broker 上,那么就将状态转为 OnlinePartition。
OfflinePartition controller 的 onBrokerFailure() 当有 broker 掉线时,将 leader 在这个机器上的 Partition 设置为 OfflinePartition
OfflinePartition TopicDeletionManager 的 completeDeleteTopic() Topic 删除成功后,中间会将该 Partition 的状态先转变为 OfflinePartition
NonExistentPartition TopicDeletionManager 的 completeDeleteTopic() Topic 删除成功后,最后会将该 Partition 的状态转移为 NonExistentPartition
NewPartition Controller 的 onNewPartitionCreation() Partition 刚创建时的一个中间状态 ,此时还没选举 leader 和设置 isr 信息

上面就是副本状态机与分区状态机的所有内容,这里只是单纯地讲述了一下这两种状态机,后续文章会开始介绍 Controller 一些其他内容,包括 Partition 迁移、Topic 新建、Topic 下线等,这些内容都会用到这篇文章讲述的内容。