diff --git a/core/src/main/scala/kafka/controller/Election.scala b/core/src/main/scala/kafka/controller/Election.scala index 1e1ee4e5b469d..450fd1d836e52 100644 --- a/core/src/main/scala/kafka/controller/Election.scala +++ b/core/src/main/scala/kafka/controller/Election.scala @@ -17,11 +17,34 @@ package kafka.controller import kafka.api.LeaderAndIsr +import kafka.common.StateChangeFailedException import org.apache.kafka.common.TopicPartition import scala.collection.Seq -case class ElectionResult(topicPartition: TopicPartition, leaderAndIsr: Option[LeaderAndIsr], liveReplicas: Seq[Int]) +sealed trait LeaderAndIsrUpdateResult { + def topicPartition: TopicPartition +} + +object LeaderAndIsrUpdateResult { + final case class Successful( + topicPartition: TopicPartition, + newLeaderAndIsr: LeaderAndIsr, + liveReplicas: Seq[Int], + replicasToStop: Seq[Int] = Seq.empty, + replicasToDelete: Seq[Int] = Seq.empty + ) extends LeaderAndIsrUpdateResult + + final case class NotNeeded( + topicPartition: TopicPartition, + currentLeaderAndIsr: LeaderAndIsr + ) extends LeaderAndIsrUpdateResult + + final case class Failed( + topicPartition: TopicPartition, + error: StateChangeFailedException + ) extends LeaderAndIsrUpdateResult +} object Election { @@ -29,7 +52,7 @@ object Election { leaderAndIsrOpt: Option[LeaderAndIsr], uncleanLeaderElectionEnabled: Boolean, isLeaderRecoverySupported: Boolean, - controllerContext: ControllerContext): ElectionResult = { + controllerContext: ControllerContext): LeaderAndIsrUpdateResult = { val assignment = controllerContext.partitionReplicaAssignment(partition) val liveReplicas = assignment.filter(replica => controllerContext.isReplicaOnline(replica, partition)) @@ -50,10 +73,19 @@ object Election { leaderAndIsr.newLeaderAndIsr(leader, newIsr) } } - ElectionResult(partition, newLeaderAndIsrOpt, liveReplicas) + newLeaderAndIsrOpt match { + case Some(newLeaderandIsr) => + LeaderAndIsrUpdateResult.Successful(partition, newLeaderandIsr, liveReplicas, Seq.empty) + case None => + LeaderAndIsrUpdateResult.Failed(partition, new StateChangeFailedException( + s"Failed to elect leader for offline partition $partition" + )) + } case None => - ElectionResult(partition, None, liveReplicas) + LeaderAndIsrUpdateResult.Failed(partition, new StateChangeFailedException( + s"Failed to elect leader for offline partition $partition" + )) } } @@ -72,49 +104,35 @@ object Election { controllerContext: ControllerContext, isLeaderRecoverySupported: Boolean, partitionsWithUncleanLeaderRecoveryState: Seq[(TopicPartition, Option[LeaderAndIsr], Boolean)] - ): Seq[ElectionResult] = { + ): Seq[LeaderAndIsrUpdateResult] = { partitionsWithUncleanLeaderRecoveryState.map { case (partition, leaderAndIsrOpt, uncleanLeaderElectionEnabled) => leaderForOffline(partition, leaderAndIsrOpt, uncleanLeaderElectionEnabled, isLeaderRecoverySupported, controllerContext) } } - private def leaderForReassign(partition: TopicPartition, - leaderAndIsr: LeaderAndIsr, - controllerContext: ControllerContext): ElectionResult = { - val targetReplicas = controllerContext.partitionFullReplicaAssignment(partition).targetReplicas - val liveReplicas = targetReplicas.filter(replica => controllerContext.isReplicaOnline(replica, partition)) - val isr = leaderAndIsr.isr - val leaderOpt = PartitionLeaderElectionAlgorithms.reassignPartitionLeaderElection(targetReplicas, isr, liveReplicas.toSet) - val newLeaderAndIsrOpt = leaderOpt.map(leader => leaderAndIsr.newLeader(leader)) - ElectionResult(partition, newLeaderAndIsrOpt, targetReplicas) - } - - /** - * Elect leaders for partitions that are undergoing reassignment. - * - * @param controllerContext Context with the current state of the cluster - * @param leaderAndIsrs A sequence of tuples representing the partitions that need election - * and their respective leader/ISR states - * - * @return The election results - */ - def leaderForReassign(controllerContext: ControllerContext, - leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)]): Seq[ElectionResult] = { - leaderAndIsrs.map { case (partition, leaderAndIsr) => - leaderForReassign(partition, leaderAndIsr, controllerContext) - } - } - private def leaderForPreferredReplica(partition: TopicPartition, leaderAndIsr: LeaderAndIsr, - controllerContext: ControllerContext): ElectionResult = { + controllerContext: ControllerContext): LeaderAndIsrUpdateResult = { val assignment = controllerContext.partitionReplicaAssignment(partition) val liveReplicas = assignment.filter(replica => controllerContext.isReplicaOnline(replica, partition)) val isr = leaderAndIsr.isr - val leaderOpt = PartitionLeaderElectionAlgorithms.preferredReplicaPartitionLeaderElection(assignment, isr, liveReplicas.toSet) - val newLeaderAndIsrOpt = leaderOpt.map(leader => leaderAndIsr.newLeader(leader)) - ElectionResult(partition, newLeaderAndIsrOpt, assignment) + if (leaderAndIsr.leader == assignment.head) { + LeaderAndIsrUpdateResult.NotNeeded(partition, leaderAndIsr) + } else { + val leaderOpt = PartitionLeaderElectionAlgorithms.preferredReplicaPartitionLeaderElection(assignment, isr, liveReplicas.toSet) + leaderOpt match { + case Some(newLeader) => + val newLeaderAndIsr = leaderAndIsr.newLeader(newLeader) + LeaderAndIsrUpdateResult.Successful(partition, newLeaderAndIsr, liveReplicas = liveReplicas) + case None => + val reason = if (!leaderAndIsr.isr.contains(assignment.head)) "offline" else "not in the ISR" + LeaderAndIsrUpdateResult.Failed(partition, new StateChangeFailedException( + s"Failed to elect preferred leader ${assignment.head} for partition $partition since " + + s"it is $reason" + )) + } + } } /** @@ -127,25 +145,88 @@ object Election { * @return The election results */ def leaderForPreferredReplica(controllerContext: ControllerContext, - leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)]): Seq[ElectionResult] = { + leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)]): Seq[LeaderAndIsrUpdateResult] = { leaderAndIsrs.map { case (partition, leaderAndIsr) => leaderForPreferredReplica(partition, leaderAndIsr, controllerContext) } } - private def leaderForControlledShutdown(partition: TopicPartition, - leaderAndIsr: LeaderAndIsr, - shuttingDownBrokerIds: Set[Int], - controllerContext: ControllerContext): ElectionResult = { + /** + * Controlled shutdown attempts to gracefully remove and fence active replicas + * that are shutting down: + * + * 1) If there is only one replica, do nothing + * 2) If the replica is the only member of the ISR, leave it as the current leader + * 3) If the replica is in the ISR, elect the next preferred leader + * 4) If the replica is not in the ISR, bump the leader epoch to fence the replica + * + * @param partition The topic partition to update + * @param leaderAndIsr The current LeaderAndIsr + * @param shuttingDownBrokerIds The brokers that are shutting down + * @param controllerContext Current controller context + * @return [[LeaderAndIsrUpdateResult.Failed]] if there is no replica available from the remaining + * replicas in the ISR after removing the shutting down replicas; or + * [[LeaderAndIsrUpdateResult.Successful]] if a new leader could be elected from the + * remaining replicas in the ISR. + */ + private def updatePartitionForControlledShutdown( + partition: TopicPartition, + leaderAndIsr: LeaderAndIsr, + shuttingDownBrokerIds: Set[Int], + controllerContext: ControllerContext + ): LeaderAndIsrUpdateResult = { + val assignment = controllerContext.partitionReplicaAssignment(partition) - val liveOrShuttingDownReplicas = assignment.filter(replica => - controllerContext.isReplicaOnline(replica, partition, includeShuttingDownBrokers = true)) + val liveReplicaIds = assignment.filter(replica => controllerContext.isReplicaOnline(replica, partition)) val isr = leaderAndIsr.isr - val leaderOpt = PartitionLeaderElectionAlgorithms.controlledShutdownPartitionLeaderElection(assignment, isr, - liveOrShuttingDownReplicas.toSet, shuttingDownBrokerIds) - val newIsr = isr.filter(replica => !shuttingDownBrokerIds.contains(replica)) - val newLeaderAndIsrOpt = leaderOpt.map(leader => leaderAndIsr.newLeaderAndIsr(leader, newIsr)) - ElectionResult(partition, newLeaderAndIsrOpt, liveOrShuttingDownReplicas) + + // Try to find a live replica in the ISR which is not shutting down. + val newLeaderIdOpt = PartitionLeaderElectionAlgorithms.controlledShutdownPartitionLeaderElection( + assignment, + isr, + liveReplicaIds.toSet, + shuttingDownBrokerIds + ) + + newLeaderIdOpt match { + case Some(newLeaderId) => + // We found a new leader from the current ISR that is not shutting down, + // so we can remove all shutting down replicas from the ISR. + val newIsr = isr.filter(replica => !shuttingDownBrokerIds.contains(replica)) + val newLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(newLeaderId, newIsr) + LeaderAndIsrUpdateResult.Successful(partition, newLeaderAndIsr, liveReplicaIds, shuttingDownBrokerIds.toSeq) + + case None => + if (leaderAndIsr.leader == LeaderAndIsr.NoLeader) { + // If there is no leader already, then we do not need to change the ISR. We + // can fence and shutdown the replicas immediately. + val newLeaderAndIsr = leaderAndIsr.newEpoch + LeaderAndIsrUpdateResult.Successful(partition, newLeaderAndIsr, liveReplicaIds, shuttingDownBrokerIds.toSeq) + } else { + // One of the shutting down replicas is the current leader. We will leave it + // as the current leader and shutdown any other replicas. + val currentLeaderId = leaderAndIsr.leader + val nonLeaderShuttingDownReplicas = shuttingDownBrokerIds.filterNot(_ == currentLeaderId) + if (nonLeaderShuttingDownReplicas.isEmpty) { + // There are no shutting down replicas besides the current leader, so make no changes. + LeaderAndIsrUpdateResult.Failed(partition, new StateChangeFailedException( + s"Failed to process controlled shutdown of broker $currentLeaderId for partition $partition " + + "since it is the leader and there are no replicas in the ISR available to elect" + )) + } else { + // We can shutdown all replicas except the current leader. We must still send + // LeaderAndIsr to the current leader that is shutting down. + val newIsr = isr.filter(replica => !nonLeaderShuttingDownReplicas.contains(replica)) + val newLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(currentLeaderId, newIsr) + LeaderAndIsrUpdateResult.Successful( + partition, + newLeaderAndIsr, + liveReplicas = liveReplicaIds :+ currentLeaderId, + replicasToStop = nonLeaderShuttingDownReplicas.toSeq + ) + } + } + } } /** @@ -157,11 +238,210 @@ object Election { * * @return The election results */ - def leaderForControlledShutdown(controllerContext: ControllerContext, - leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)]): Seq[ElectionResult] = { + def processControlledShutdown( + controllerContext: ControllerContext, + leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)] + ): Seq[LeaderAndIsrUpdateResult] = { val shuttingDownBrokerIds = controllerContext.shuttingDownBrokerIds.toSet leaderAndIsrs.map { case (partition, leaderAndIsr) => - leaderForControlledShutdown(partition, leaderAndIsr, shuttingDownBrokerIds, controllerContext) + updatePartitionForControlledShutdown(partition, leaderAndIsr, shuttingDownBrokerIds, controllerContext) } } + + def processReassignmentCancellation( + controllerContext: ControllerContext, + leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)] + ): Seq[LeaderAndIsrUpdateResult] = { + leaderAndIsrs.map { case (topicPartition, leaderAndIsr) => + val assignment = controllerContext.partitionFullReplicaAssignment(topicPartition) + if (!assignment.isBeingReassigned) { + throw new IllegalStateException(s"Cancellation of reassignment failed for $topicPartition " + + "since there is not reassignment in progress") + } + + updateLeaderAndIsrForCancelledReassignment( + topicPartition, + controllerContext, + assignment, + leaderAndIsr + ) + } + } + + /** + * Build a new LeaderAndIsr state in order to cancel an active reassignment. This is done + * by removing the `AddingReplicas` set from the ISR and electing a new leader from the + * original replicas (if necessary). + * + * Note that if the `AddingReplicas` set is empty (i.e. if the reassignment is removing replicas + * only), then no changes to the LeaderAndIsr state will be made. On the other hand, if this + * set is not empty, then this method guarantees a bump to the leader epoch. This ensures that + * a `StopReplica` request can be sent to any replicas that need to be deleted with a larger + * epoch than previously seen. + * + * An error will be returned if a leader cannot be elected from the original replica set. + * + * @param topicPartition The topic partition to update + * @param controllerContext Current controller context + * @param assignment The current reassignment which is being cancelled + * @param leaderAndIsr The current LeaderAndIsr state + * @return [[LeaderAndIsrUpdateResult.NotNeeded]] if `AddingReplicas` is empty; + * [[LeaderAndIsrUpdateResult.Failed]] if no live leader from the original replicas could be elected; or + * [[LeaderAndIsrUpdateResult.Successful]] if all `AddingReplicas` were successfully removed from the + * ISR and a new leader elected + */ + private def updateLeaderAndIsrForCancelledReassignment( + topicPartition: TopicPartition, + controllerContext: ControllerContext, + assignment: ReplicaAssignment, + leaderAndIsr: LeaderAndIsr + ): LeaderAndIsrUpdateResult = { + val addingReplicas = assignment.addingReplicas + if (addingReplicas.isEmpty) { + // If there are no adding replicas, then there are no partitions to stop and delete + LeaderAndIsrUpdateResult.NotNeeded(topicPartition, leaderAndIsr) + } else { + // Otherwise, we need to remove the adding replicas and maybe elect a new leader + val originReplicas = assignment.originReplicas + val liveOriginReplicas = originReplicas.filter(replica => controllerContext.isReplicaOnline(replica, topicPartition)) + val newIsr = leaderAndIsr.isr.filter(originReplicas.contains) + val currentLeader = leaderAndIsr.leader + + if (newIsr.isEmpty) { + // If there are no replicas from the original set in the ISR, then it is not possible + // to cancel the assignment without having an unclean election. + LeaderAndIsrUpdateResult.Failed(topicPartition, new StateChangeFailedException( + s"Failed to cancel reassignment $assignment for partition $topicPartition since " + + s"there are no replicas in the original replica set $originReplicas remaining in the ISR" + )) + } else { + val newLeaderOpt = if (originReplicas.contains(currentLeader)) { + // The leader is already among the origin replicas, so we can keep it. + Some(currentLeader) + } else { + // Try to use the preferred origin leader if it is live and in the ISR + val preferredOriginLeader = originReplicas.head + if (liveOriginReplicas.contains(preferredOriginLeader) && newIsr.contains(preferredOriginLeader)) { + Some(preferredOriginLeader) + } else { + // Otherwise, use a random live replica from the original set if available + newIsr.find(liveOriginReplicas.contains) + } + } + + newLeaderOpt match { + case Some(newLeader) => + val newLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(newLeader, newIsr) + LeaderAndIsrUpdateResult.Successful( + topicPartition, + newLeaderAndIsr, + liveReplicas = liveOriginReplicas, + replicasToDelete = addingReplicas + ) + case None => + LeaderAndIsrUpdateResult.Failed(topicPartition, new StateChangeFailedException( + s"Failed to elect new leader for $topicPartition from the live original replicas: $liveOriginReplicas" + )) + } + } + } + } + + def processReassignmentCompletion( + controllerContext: ControllerContext, + leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)] + ): Seq[LeaderAndIsrUpdateResult] = { + leaderAndIsrs.map { case (topicPartition, leaderAndIsr) => + val assignment = controllerContext.partitionFullReplicaAssignment(topicPartition) + if (!assignment.isBeingReassigned) { + throw new IllegalStateException(s"Completion of reassignment failed for $topicPartition " + + "since there is not reassignment in progress") + } + + updateLeaderAndIsrForCompletedReassignment( + topicPartition, + controllerContext, + assignment, + leaderAndIsr + ) + } + } + + /** + * Build a new LeaderAndIsr state in order to complete an active reassignment. This is done + * by removing the `RemovingReplicas` set from the ISR and electing a new leader from the + * target replicas (if necessary). + * + * Note that if the `RemovingReplicas` set is empty (i.e. if the reassignment is adding replicas + * only), then no changes to the LeaderAndIsr state will be made. On the other hand, if this + * set is not empty, then this method guarantees a bump to the leader epoch. This ensures that + * a `StopReplica` request can be sent to any replicas that need to be deleted with a larger + * epoch than previously seen. + * + * An error will be returned if a leader cannot be elected from the original replica set. + * + * @param topicPartition The topic partition to update + * @param controllerContext Current controller context + * @param assignment The current reassignment which is being completed + * @param leaderAndIsr The current LeaderAndIsr state + * @return [[LeaderAndIsrUpdateResult.NotNeeded]] if `RemovingReplicas` is empty; + * [[LeaderAndIsrUpdateResult.Failed]] if there are any target replicas NOT in the ISR or if + * no live leader from the target replicas could be elected; or + * [[LeaderAndIsrUpdateResult.Successful]] if all `RemovingReplicas` were successfully removed from the + * ISR and a new leader elected + */ + private def updateLeaderAndIsrForCompletedReassignment( + topicPartition: TopicPartition, + controllerContext: ControllerContext, + assignment: ReplicaAssignment, + leaderAndIsr: LeaderAndIsr + ): LeaderAndIsrUpdateResult = { + val removingReplicas = assignment.removingReplicas + val targetReplicas = assignment.targetReplicas + val newIsr = leaderAndIsr.isr.filter(targetReplicas.contains) + + // For a reassignment to be completed, all of the target replicas must be in the ISR. + if (targetReplicas.size != newIsr.size) { + LeaderAndIsrUpdateResult.Failed(topicPartition, new StateChangeFailedException( + s"Failed to complete reassignment for partition $topicPartition since the current ISR " + + s"${leaderAndIsr.isr} does not contain all of the target replicas $targetReplicas" + )) + } else if (removingReplicas.isEmpty) { + // There are no replicas to remove, so there is no need to update the Leader and ISR + LeaderAndIsrUpdateResult.NotNeeded(topicPartition, leaderAndIsr) + } else { + val liveTargetReplicas = targetReplicas.filter(replica => controllerContext.isReplicaOnline(replica, topicPartition)) + val currentLeader = leaderAndIsr.leader + + val newLeaderOpt = if (targetReplicas.contains(currentLeader)) { + // The current leader is already among target replicas, so we can keep it + Some(currentLeader) + } else { + // Try to use the preferred target leader if it is live and in the ISR + val preferredTargetLeader = targetReplicas.head + if (liveTargetReplicas.contains(preferredTargetLeader) && newIsr.contains(preferredTargetLeader)) { + Some(preferredTargetLeader) + } else { + // Otherwise choose a random live replica from the ISR + newIsr.find(liveTargetReplicas.contains) + } + } + + newLeaderOpt match { + case Some(newLeader) => + val newLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(newLeader, newIsr) + LeaderAndIsrUpdateResult.Successful( + topicPartition, + newLeaderAndIsr, + liveReplicas = liveTargetReplicas, + replicasToDelete = removingReplicas + ) + case None => + LeaderAndIsrUpdateResult.Failed(topicPartition, new StateChangeFailedException( + s"Failed to elect new leader for $topicPartition from the live target replicas: $liveTargetReplicas" + )) + } + } + } + } diff --git a/core/src/main/scala/kafka/controller/KafkaController.scala b/core/src/main/scala/kafka/controller/KafkaController.scala index 289de9ab29730..d33ba80933239 100644 --- a/core/src/main/scala/kafka/controller/KafkaController.scala +++ b/core/src/main/scala/kafka/controller/KafkaController.scala @@ -654,7 +654,7 @@ class KafkaController(val config: KafkaConfig, partitionStateMachine.handleStateChanges( newPartitions.toSeq, OnlinePartition, - Some(OfflinePartitionLeaderElectionStrategy(false)) + Some(OfflineLeaderElectionStrategy(false)) ) replicaStateMachine.handleStateChanges(controllerContext.replicasForPartition(newPartitions).toSeq, OnlineReplica) } @@ -696,19 +696,13 @@ class KafkaController(val config: KafkaConfig, * Phase B (when TRS = ISR): The reassignment is complete * * B1. Move all replicas in AR to OnlineReplica state. - * B2. Set RS = TRS, AR = [], RR = [] in memory. - * B3. Send a LeaderAndIsr request with RS = TRS. This will prevent the leader from adding any replica in TRS - ORS back in the isr. - * If the current leader is not in TRS or isn't alive, we move the leader to a new replica in TRS. - * We may send the LeaderAndIsr to more than the TRS replicas due to the - * way the partition state machine works (it reads replicas from ZK) - * B4. Move all replicas in RR to OfflineReplica state. As part of OfflineReplica state change, we shrink the - * isr to remove RR in ZooKeeper and send a LeaderAndIsr ONLY to the Leader to notify it of the shrunk isr. - * After that, we send a StopReplica (delete = false) to the replicas in RR. - * B5. Move all replicas in RR to NonExistentReplica state. This will send a StopReplica (delete = true) to - * the replicas in RR to physically delete the replicas on disk. - * B6. Update ZK with RS=TRS, AR=[], RR=[]. - * B7. Remove the ISR reassign listener and maybe update the /admin/reassign_partitions path in ZK to remove this partition from it if present. - * B8. After electing leader, the replicas and isr information changes. So resend the update metadata request to every broker. + * B2. Remove any replicas from RR from the ISR and ensure a leader is elected from TRS. Send + * LeaderAndIsr to all TRS and StopReplica(delete=true) to all replicas in RR. + * B3. Transition all replicas in RR to `UnassignedReplica` which removes their replica states. + * B4. Set RS = TRS, AR = [], RR = [] in memory. + * B5. Update ZK with RS=TRS, AR=[], RR=[]. + * B6. Remove the ISR reassign listener and maybe update the /admin/reassign_partitions path in ZK to remove this partition from it if present. + * B7. After electing leader, the replicas and isr information changes. So resend the update metadata request to every broker. * * In general, there are two goals we want to aim for: * 1. Every replica present in the replica set of a LeaderAndIsrRequest gets the request sent to it @@ -727,40 +721,46 @@ class KafkaController(val config: KafkaConfig, * Note that we have to update RS in ZK with TRS last since it's the only place where we store ORS persistently. * This way, if the controller crashes before that step, we can still recover. */ - private def onPartitionReassignment(topicPartition: TopicPartition, reassignment: ReplicaAssignment): Unit = { - // While a reassignment is in progress, deletion is not allowed - topicDeletionManager.markTopicIneligibleForDeletion(Set(topicPartition.topic), reason = "topic reassignment in progress") + private def onPartitionReassignment(topicPartition: TopicPartition, assignment: ReplicaAssignment): Unit = { + maybeUpdateCurrentAssignment(topicPartition, assignment) - updateCurrentReassignment(topicPartition, reassignment) + if (!assignment.isBeingReassigned) { + // If a reassignment is cancelled or if a new reassignment is just changing the + // preferred leader, then it can be completed without additional replica movements. + // We do still need to send `UpdateMetadata` though for the replica changes. + removePartitionFromReassigningPartitions(topicPartition, assignment) + sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(topicPartition)) + return + } - val addingReplicas = reassignment.addingReplicas - val removingReplicas = reassignment.removingReplicas + val addingReplicas = assignment.addingReplicas + val removingReplicas = assignment.removingReplicas - if (!isReassignmentComplete(topicPartition, reassignment)) { + if (!isReassignmentComplete(topicPartition, assignment)) { + addPartitionToReassigningPartitions(topicPartition) // A1. Send LeaderAndIsr request to every replica in ORS + TRS (with the new RS, AR and RR). - updateLeaderEpochAndSendRequest(topicPartition, reassignment) + updateLeaderEpochAndSendRequest(topicPartition, assignment) // A2. replicas in AR -> NewReplica startNewReplicasForReassignedPartition(topicPartition, addingReplicas) } else { // B1. replicas in AR -> OnlineReplica replicaStateMachine.handleStateChanges(addingReplicas.map(PartitionAndReplica(topicPartition, _)), OnlineReplica) - // B2. Set RS = TRS, AR = [], RR = [] in memory. - val completedReassignment = ReplicaAssignment(reassignment.targetReplicas) - controllerContext.updatePartitionFullReplicaAssignment(topicPartition, completedReassignment) - // B3. Send LeaderAndIsr request with a potential new leader (if current leader not in TRS) and - // a new RS (using TRS) and same isr to every broker in ORS + TRS or TRS - moveReassignedPartitionLeaderIfRequired(topicPartition, completedReassignment) - // B4. replicas in RR -> Offline (force those replicas out of isr) - // B5. replicas in RR -> NonExistentReplica (force those replicas to be deleted) + // B2. Send LeaderAndIsr request with a potential new leader (if current leader not in TRS) and + // StopReplica(delete=true) to all replicas being removed + partitionStateMachine.handleStateChanges(Seq(topicPartition), OnlinePartition, Some(CompleteReassignmentStrategy)) + // B3. Transition all replicas in RR to `UnassignedReplica` so that replica state is deleted stopRemovedReplicasOfReassignedPartition(topicPartition, removingReplicas) - // B6. Update ZK with RS = TRS, AR = [], RR = []. + // B4. Set RS = TRS, AR = [], RR = [] in memory. + val completedReassignment = ReplicaAssignment(assignment.targetReplicas) + info(s"Updating assignment of $topicPartition to $completedReassignment following completion " + + s"of reassignment $assignment") + controllerContext.updatePartitionFullReplicaAssignment(topicPartition, completedReassignment) + // B5. Update ZK with RS = TRS, AR = [], RR = []. updateReplicaAssignmentForPartition(topicPartition, completedReassignment) - // B7. Remove the ISR reassign listener and maybe update the /admin/reassign_partitions path in ZK to remove this partition from it. + // B6. Remove the ISR reassign listener and maybe update the /admin/reassign_partitions path in ZK to remove this partition from it. removePartitionFromReassigningPartitions(topicPartition, completedReassignment) - // B8. After electing a leader in B3, the replicas and isr information changes, so resend the update metadata request to every broker + // B7. After electing a leader in B2, the replicas and isr information changes, so resend the update metadata request to every broker sendUpdateMetadataRequest(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(topicPartition)) - // signal delete topic thread if reassignment for some partitions belonging to topics being deleted just completed - topicDeletionManager.resumeDeletionForTopics(Set(topicPartition.topic)) } } @@ -776,33 +776,60 @@ class KafkaController(val config: KafkaConfig, * will be encoded as [3, 4, 2, 1] while the reassignment is in progress. If the reassignment * is cancelled, there is no way to restore the original order. * - * @param topicPartition The reassigning partition - * @param reassignment The new reassignment + * @param topicPartition The topic partition + * @param newAssignment The new assignment */ - private def updateCurrentReassignment(topicPartition: TopicPartition, reassignment: ReplicaAssignment): Unit = { + private def maybeUpdateCurrentAssignment(topicPartition: TopicPartition, newAssignment: ReplicaAssignment): Unit = { val currentAssignment = controllerContext.partitionFullReplicaAssignment(topicPartition) + if (currentAssignment != newAssignment) { + if (currentAssignment.isBeingReassigned) { + // Cancel the current reassignment by removing unneeded replicas from the ISR + // and stopping/deleting them. Note that if the controller fails before updating + // the assignment state in Zookeeper below, these replicas may get restarted after + // controller fail-over. We expect the client would retry the cancellation in this case. + cancelReassignment(topicPartition, currentAssignment, newAssignment) + } - if (currentAssignment != reassignment) { - debug(s"Updating assignment of partition $topicPartition from $currentAssignment to $reassignment") - + info(s"Updating assignment of partition $topicPartition from $currentAssignment to $newAssignment") // U1. Update assignment state in zookeeper - updateReplicaAssignmentForPartition(topicPartition, reassignment) + updateReplicaAssignmentForPartition(topicPartition, newAssignment) // U2. Update assignment state in memory - controllerContext.updatePartitionFullReplicaAssignment(topicPartition, reassignment) - - // If there is a reassignment already in progress, then some of the currently adding replicas - // may be eligible for immediate removal, in which case we need to stop the replicas. - val unneededReplicas = currentAssignment.replicas.diff(reassignment.replicas) - if (unneededReplicas.nonEmpty) - stopRemovedReplicasOfReassignedPartition(topicPartition, unneededReplicas) + controllerContext.updatePartitionFullReplicaAssignment(topicPartition, newAssignment) } + } - if (!isAlterPartitionEnabled) { - val reassignIsrChangeHandler = new PartitionReassignmentIsrChangeHandler(eventManager, topicPartition) - zkClient.registerZNodeChangeHandler(reassignIsrChangeHandler) + private def cancelReassignment( + topicPartition: TopicPartition, + oldAssignment: ReplicaAssignment, + newAssignment: ReplicaAssignment, + ): Unit = { + info(s"Cancelling active reassignment $oldAssignment for $topicPartition") + + // First we need to remove any unneeded replicas from the ISR and update the leader + // if necessary. This will ensure that `StopReplica` requests will be sent to the + // unneeded replicas, but they will not be deleted yet. + val results = partitionStateMachine.handleStateChanges( + Seq(topicPartition), + OnlinePartition, + Some(CancelReassignmentStrategy) + ) + + val result: Either[Throwable, LeaderAndIsr] = results.getOrElse(topicPartition, + throw new IllegalStateException(s"Partition $topicPartition not included in result map " + + s"after attempting to change state to $OnlinePartition using $CancelReassignmentStrategy") + ) + + result.left.exists { exception => + throw new KafkaException("Failed to cancel current reassignment " + + s"$oldAssignment for partition $topicPartition", exception) } - controllerContext.partitionsBeingReassigned.add(topicPartition) + // We have already stopped and deleted the unneeded replicas, but we also need to + // remove their associated states. + val unneededReplicas = oldAssignment.replicas.diff(newAssignment.replicas) + if (unneededReplicas.nonEmpty) { + stopRemovedReplicasOfReassignedPartition(topicPartition, unneededReplicas) + } } /** @@ -821,9 +848,7 @@ class KafkaController(val config: KafkaConfig, */ private def maybeTriggerPartitionReassignment(reassignments: Map[TopicPartition, ReplicaAssignment]): Map[TopicPartition, ApiError] = { reassignments.map { case (tp, reassignment) => - val topic = tp.topic - - val apiError = if (topicDeletionManager.isTopicQueuedUpForDeletion(topic)) { + val apiError = if (topicDeletionManager.isTopicQueuedUpForDeletion(tp.topic)) { info(s"Skipping reassignment of $tp since the topic is currently being deleted") new ApiError(Errors.UNKNOWN_TOPIC_OR_PARTITION, "The partition does not exist.") } else { @@ -834,7 +859,6 @@ class KafkaController(val config: KafkaConfig, ApiError.NONE } catch { case e: ControllerMovedException => - info(s"Failed completing reassignment of partition $tp because controller has moved to another broker") throw e case e: Throwable => error(s"Error completing reassignment of partition $tp", e) @@ -865,12 +889,12 @@ class KafkaController(val config: KafkaConfig, info(s"Starting replica leader election ($electionType) for partitions ${partitions.mkString(",")} triggered by $electionTrigger") try { val strategy = electionType match { - case ElectionType.PREFERRED => PreferredReplicaPartitionLeaderElectionStrategy + case ElectionType.PREFERRED => PreferredLeaderElectionStrategy case ElectionType.UNCLEAN => /* Let's be conservative and only trigger unclean election if the election type is unclean and it was * triggered by the admin client */ - OfflinePartitionLeaderElectionStrategy(allowUnclean = electionTrigger == AdminClientTriggered) + OfflineLeaderElectionStrategy(allowUnclean = electionTrigger == AdminClientTriggered) } val results = partitionStateMachine.handleStateChanges( @@ -999,38 +1023,13 @@ class KafkaController(val config: KafkaConfig, } } - private def moveReassignedPartitionLeaderIfRequired(topicPartition: TopicPartition, - newAssignment: ReplicaAssignment): Unit = { - val reassignedReplicas = newAssignment.replicas - val currentLeader = controllerContext.partitionLeadershipInfo(topicPartition).get.leaderAndIsr.leader - - if (!reassignedReplicas.contains(currentLeader)) { - info(s"Leader $currentLeader for partition $topicPartition being reassigned, " + - s"is not in the new list of replicas ${reassignedReplicas.mkString(",")}. Re-electing leader") - // move the leader to one of the alive and caught up new replicas - partitionStateMachine.handleStateChanges(Seq(topicPartition), OnlinePartition, Some(ReassignPartitionLeaderElectionStrategy)) - } else if (controllerContext.isReplicaOnline(currentLeader, topicPartition)) { - info(s"Leader $currentLeader for partition $topicPartition being reassigned, " + - s"is already in the new list of replicas ${reassignedReplicas.mkString(",")} and is alive") - // shrink replication factor and update the leader epoch in zookeeper to use on the next LeaderAndIsrRequest - updateLeaderEpochAndSendRequest(topicPartition, newAssignment) - } else { - info(s"Leader $currentLeader for partition $topicPartition being reassigned, " + - s"is already in the new list of replicas ${reassignedReplicas.mkString(",")} but is dead") - partitionStateMachine.handleStateChanges(Seq(topicPartition), OnlinePartition, Some(ReassignPartitionLeaderElectionStrategy)) - } - } - private def stopRemovedReplicasOfReassignedPartition(topicPartition: TopicPartition, removedReplicas: Seq[Int]): Unit = { - // first move the replica to offline state (the controller removes it from the ISR) + // Replicas are stopped and deleted as part of the `Online` partition transition + // using the reassignment cancellation or completion strategy. The transition to + // `UnassignedReplica` here just cleans up the replica state. val replicasToBeDeleted = removedReplicas.map(PartitionAndReplica(topicPartition, _)) - replicaStateMachine.handleStateChanges(replicasToBeDeleted, OfflineReplica) - // send stop replica command to the old replicas - replicaStateMachine.handleStateChanges(replicasToBeDeleted, ReplicaDeletionStarted) - // TODO: Eventually partition reassignment could use a callback that does retries if deletion failed - replicaStateMachine.handleStateChanges(replicasToBeDeleted, ReplicaDeletionSuccessful) - replicaStateMachine.handleStateChanges(replicasToBeDeleted, NonExistentReplica) + replicaStateMachine.handleStateChanges(replicasToBeDeleted, UnassignedReplica) } private def updateReplicaAssignmentForPartition(topicPartition: TopicPartition, assignment: ReplicaAssignment): Unit = { @@ -1113,18 +1112,29 @@ class KafkaController(val config: KafkaConfig, } } + private def addPartitionToReassigningPartitions( + topicPartition: TopicPartition, + ): Unit = { + // While a reassignment is in progress, deletion is not allowed + topicDeletionManager.markTopicIneligibleForDeletion(Set(topicPartition.topic), reason = "topic reassignment in progress") + + if (!isAlterPartitionEnabled) { + val reassignIsrChangeHandler = new PartitionReassignmentIsrChangeHandler(eventManager, topicPartition) + zkClient.registerZNodeChangeHandler(reassignIsrChangeHandler) + } + controllerContext.partitionsBeingReassigned.add(topicPartition) + } + private def removePartitionFromReassigningPartitions(topicPartition: TopicPartition, assignment: ReplicaAssignment): Unit = { - if (controllerContext.partitionsBeingReassigned.contains(topicPartition)) { - if (!isAlterPartitionEnabled) { - val path = TopicPartitionStateZNode.path(topicPartition) - zkClient.unregisterZNodeChangeHandler(path) - } - maybeRemoveFromZkReassignment((tp, replicas) => tp == topicPartition && replicas == assignment.replicas) - controllerContext.partitionsBeingReassigned.remove(topicPartition) - } else { - throw new IllegalStateException("Cannot remove a reassigning partition because it is not present in memory") + if (!isAlterPartitionEnabled) { + val path = TopicPartitionStateZNode.path(topicPartition) + zkClient.unregisterZNodeChangeHandler(path) } + maybeRemoveFromZkReassignment((tp, replicas) => tp == topicPartition && replicas == assignment.replicas) + controllerContext.partitionsBeingReassigned.remove(topicPartition) + // signal delete topic thread if reassignment for some partitions belonging to topics being deleted just completed + topicDeletionManager.resumeDeletionForTopics(Set(topicPartition.topic)) } /** @@ -1311,7 +1321,7 @@ class KafkaController(val config: KafkaConfig, controlledShutdownCallback(controlledShutdownResult) } - private def doControlledShutdown(id: Int, brokerEpoch: Long): Set[TopicPartition] = { + private def doControlledShutdown(brokerId: Int, brokerEpoch: Long): Set[TopicPartition] = { if (!isActive) { throw new ControllerMovedException("Controller moved to another broker. Aborting controlled shutdown") } @@ -1319,48 +1329,43 @@ class KafkaController(val config: KafkaConfig, // broker epoch in the request is unknown if the controller hasn't been upgraded to use KIP-380 // so we will keep the previous behavior and don't reject the request if (brokerEpoch != AbstractControlRequest.UNKNOWN_BROKER_EPOCH) { - val cachedBrokerEpoch = controllerContext.liveBrokerIdAndEpochs(id) + val cachedBrokerEpoch = controllerContext.liveBrokerIdAndEpochs(brokerId) if (brokerEpoch < cachedBrokerEpoch) { val stateBrokerEpochErrorMessage = "Received controlled shutdown request from an old broker epoch " + - s"$brokerEpoch for broker $id. Current broker epoch is $cachedBrokerEpoch." + s"$brokerEpoch for broker $brokerId. Current broker epoch is $cachedBrokerEpoch." info(stateBrokerEpochErrorMessage) throw new StaleBrokerEpochException(stateBrokerEpochErrorMessage) } } - info(s"Shutting down broker $id") + info(s"Shutting down broker $brokerId") - if (!controllerContext.liveOrShuttingDownBrokerIds.contains(id)) - throw new BrokerNotAvailableException(s"Broker id $id does not exist.") + if (!controllerContext.liveOrShuttingDownBrokerIds.contains(brokerId)) + throw new BrokerNotAvailableException(s"Broker id $brokerId does not exist.") - controllerContext.shuttingDownBrokerIds.add(id) + controllerContext.shuttingDownBrokerIds.add(brokerId) debug(s"All shutting down brokers: ${controllerContext.shuttingDownBrokerIds.mkString(",")}") debug(s"Live brokers: ${controllerContext.liveBrokerIds.mkString(",")}") - val partitionsToActOn = controllerContext.partitionsOnBroker(id).filter { partition => - controllerContext.partitionReplicaAssignment(partition).size > 1 && - controllerContext.partitionLeadershipInfo(partition).isDefined && + val partitionsToActOn = controllerContext.partitionsOnBroker(brokerId).filter { partition => + controllerContext.partitionLeadershipInfo(partition).isDefined && !topicDeletionManager.isTopicQueuedUpForDeletion(partition.topic) } - val (partitionsLedByBroker, partitionsFollowedByBroker) = partitionsToActOn.partition { partition => - controllerContext.partitionLeadershipInfo(partition).get.leaderAndIsr.leader == id - } - partitionStateMachine.handleStateChanges(partitionsLedByBroker.toSeq, OnlinePartition, Some(ControlledShutdownPartitionLeaderElectionStrategy)) - try { - brokerRequestBatch.newBatch() - partitionsFollowedByBroker.foreach { partition => - brokerRequestBatch.addStopReplicaRequestForBrokers(Seq(id), partition, deletePartition = false) - } - brokerRequestBatch.sendRequestsToBrokers(epoch) - } catch { - case e: IllegalStateException => - handleIllegalState(e) - } - // If the broker is a follower, updates the isr in ZK and notifies the current leader - replicaStateMachine.handleStateChanges(partitionsFollowedByBroker.map(partition => - PartitionAndReplica(partition, id)).toSeq, OfflineReplica) - trace(s"All leaders = ${controllerContext.partitionsLeadershipInfo.mkString(",")}") - controllerContext.partitionLeadersOnBroker(id) + + // Attempt to elect new leaders if needed and remove the broker from ISRs. + // This will bump the epoch and send `StopReplica` to the shutting down broker + // for any partition that it is no longer the leader of. + partitionStateMachine.handleStateChanges( + partitionsToActOn.toSeq, + OnlinePartition, + Some(ControlledShutdownStrategy) + ) + + val remainingLeaderPartitions = controllerContext.partitionLeadersOnBroker(brokerId) + info(s"After attempting controlled shutdown of broker $brokerId, " + + s"there are ${remainingLeaderPartitions.size} including " + + s"${remainingLeaderPartitions.take(5)}") + remainingLeaderPartitions } private def processUpdateMetadataResponseReceived(updateMetadataResponse: UpdateMetadataResponse, brokerId: Int): Unit = { @@ -1675,7 +1680,7 @@ class KafkaController(val config: KafkaConfig, updatedTopicIdAssignments.foreach { topicIdAssignment => topicIdAssignment.topicId.foreach { topicId => controllerContext.addTopicId(topicIdAssignment.topic, topicId) - } + } } } @@ -1775,10 +1780,8 @@ class KafkaController(val config: KafkaConfig, val partitionsToReassign = mutable.Map.empty[TopicPartition, ReplicaAssignment] zkClient.getPartitionReassignment.forKeyValue { (tp, targetReplicas) => - maybeBuildReassignment(tp, Some(targetReplicas)) match { - case Some(context) => partitionsToReassign.put(tp, context) - case None => reassignmentResults.put(tp, new ApiError(Errors.NO_REASSIGNMENT_IN_PROGRESS)) - } + val currentAssignment = controllerContext.partitionFullReplicaAssignment(tp) + partitionsToReassign.put(tp, currentAssignment.reassignTo(targetReplicas)) } reassignmentResults ++= maybeTriggerPartitionReassignment(partitionsToReassign) @@ -1811,61 +1814,92 @@ class KafkaController(val config: KafkaConfig, val reassignmentResults = mutable.Map.empty[TopicPartition, ApiError] val partitionsToReassign = mutable.Map.empty[TopicPartition, ReplicaAssignment] - reassignments.forKeyValue { (tp, targetReplicas) => - val maybeApiError = targetReplicas.flatMap(validateReplicas(tp, _)) - maybeApiError match { + reassignments.forKeyValue { (tp, targetReplicasOpt) => + val currentAssignment = controllerContext.partitionFullReplicaAssignment(tp) + targetReplicasOpt match { + case Some(targetReplicas) => + validateReassignmentTargetReplicas(tp, currentAssignment, targetReplicas) match { + case Left(err) => reassignmentResults.put(tp, err) + case Right(newAssignment) => partitionsToReassign.put(tp, newAssignment) + } + case None => - maybeBuildReassignment(tp, targetReplicas) match { - case Some(context) => partitionsToReassign.put(tp, context) - case None => reassignmentResults.put(tp, new ApiError(Errors.NO_REASSIGNMENT_IN_PROGRESS)) + // This is a cancellation request + if (!currentAssignment.isBeingReassigned) { + reassignmentResults.put(tp, new ApiError(Errors.NO_REASSIGNMENT_IN_PROGRESS)) + } else { + validateReassignmentCancellation(tp, currentAssignment) match { + case Left(error) => reassignmentResults.put(tp, error) + case Right(newAssignment) => partitionsToReassign.put(tp, newAssignment) + } } - case Some(err) => - reassignmentResults.put(tp, err) } } - // The latest reassignment (whether by API or through zk) always takes precedence, - // so remove from active zk reassignment (if one exists) - maybeRemoveFromZkReassignment((tp, _) => partitionsToReassign.contains(tp)) + try { + // The latest reassignment (whether by API or through zk) always takes precedence, + // so remove from active zk reassignment (if one exists) + maybeRemoveFromZkReassignment((tp, _) => partitionsToReassign.contains(tp)) - reassignmentResults ++= maybeTriggerPartitionReassignment(partitionsToReassign) - callback(Left(reassignmentResults)) + reassignmentResults ++= maybeTriggerPartitionReassignment(partitionsToReassign) + callback(Left(reassignmentResults)) + } catch { + case e: ControllerMovedException => + info(s"Failed to complete reassignment of ${partitionsToReassign.size} partitions because " + + s"the controller has moved to another broker") + callback(Right(new ApiError(Errors.NOT_CONTROLLER))) + throw e + } + } + } + + private def validateReassignmentCancellation( + topicPartition: TopicPartition, + currentAssignment: ReplicaAssignment + ): Either[ApiError, ReplicaAssignment] = { + controllerContext.partitionsLeadershipInfo.get(topicPartition) match { + case Some(leadershipInfo) => + // If we are canceling a reassignment, we need to verify that some of the original + // replicas are still in the ISR. + if (leadershipInfo.leaderAndIsr.isr.intersect(currentAssignment.originReplicas).isEmpty) { + Left(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, "Cannot cancel current " + + s"reassignment $currentAssignment because none of the original replicas are in the " + + s"current ISR ${leadershipInfo.leaderAndIsr.isr}")) + } else { + val newAssignment = currentAssignment.reassignTo(currentAssignment.originReplicas) + Right(newAssignment) + } + + case None => + Left(new ApiError(Errors.UNKNOWN_TOPIC_OR_PARTITION)) } } - private def validateReplicas(topicPartition: TopicPartition, replicas: Seq[Int]): Option[ApiError] = { + private def validateReassignmentTargetReplicas( + topicPartition: TopicPartition, + currentAssignment: ReplicaAssignment, + replicas: Seq[Int] + ): Either[ApiError, ReplicaAssignment] = { val replicaSet = replicas.toSet - if (replicas.isEmpty) - Some(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, + if (replicas.isEmpty) { + Left(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, s"Empty replica list specified in partition reassignment.")) - else if (replicas.size != replicaSet.size) { - Some(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, + } else if (replicas.size != replicaSet.size) { + Left(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, s"Duplicate replica ids in partition reassignment replica list: $replicas")) - } else if (replicas.exists(_ < 0)) - Some(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, + } else if (replicas.exists(_ < 0)) { + Left(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, s"Invalid broker id in replica list: $replicas")) - else { + } else { // Ensure that any new replicas are among the live brokers - val currentAssignment = controllerContext.partitionFullReplicaAssignment(topicPartition) val newAssignment = currentAssignment.reassignTo(replicas) val areNewReplicasAlive = newAssignment.addingReplicas.toSet.subsetOf(controllerContext.liveBrokerIds) - if (!areNewReplicasAlive) - Some(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, + if (!areNewReplicasAlive) { + Left(new ApiError(Errors.INVALID_REPLICA_ASSIGNMENT, s"Replica assignment has brokers that are not alive. Replica list: " + s"${newAssignment.addingReplicas}, live broker list: ${controllerContext.liveBrokerIds}")) - else None - } - } - - private def maybeBuildReassignment(topicPartition: TopicPartition, - targetReplicasOpt: Option[Seq[Int]]): Option[ReplicaAssignment] = { - val replicaAssignment = controllerContext.partitionFullReplicaAssignment(topicPartition) - if (replicaAssignment.isBeingReassigned) { - val targetReplicas = targetReplicasOpt.getOrElse(replicaAssignment.originReplicas) - Some(replicaAssignment.reassignTo(targetReplicas)) - } else { - targetReplicasOpt.map { targetReplicas => - replicaAssignment.reassignTo(targetReplicas) + } else { + Right(newAssignment) } } } @@ -2653,7 +2687,7 @@ case class PartitionAndReplica(topicPartition: TopicPartition, replica: Int) { def partition: Int = topicPartition.partition override def toString: String = { - s"[Topic=$topic,Partition=$partition,Replica=$replica]" + s"[TopicPartition=$topicPartition,Replica=$replica]" } } diff --git a/core/src/main/scala/kafka/controller/PartitionStateMachine.scala b/core/src/main/scala/kafka/controller/PartitionStateMachine.scala index 71b163a2e2243..e8c4a5a5f5753 100755 --- a/core/src/main/scala/kafka/controller/PartitionStateMachine.scala +++ b/core/src/main/scala/kafka/controller/PartitionStateMachine.scala @@ -73,7 +73,7 @@ abstract class PartitionStateMachine(controllerContext: ControllerContext) exten !controllerContext.isTopicQueuedUpForDeletion(partition.topic) }.toSeq - handleStateChanges(partitionsToTrigger, OnlinePartition, Some(OfflinePartitionLeaderElectionStrategy(false))) + handleStateChanges(partitionsToTrigger, OnlinePartition, Some(OfflineLeaderElectionStrategy(false))) // TODO: If handleStateChanges catches an exception, it is not enough to bail out and log an error. // It is important to trigger leader election for those partitions. } @@ -109,7 +109,7 @@ abstract class PartitionStateMachine(controllerContext: ControllerContext) exten def handleStateChanges( partitions: Seq[TopicPartition], targetState: PartitionState, - leaderElectionStrategy: Option[PartitionLeaderElectionStrategy] + leaderElectionStrategy: Option[LeaderAndIsrUpdateStrategy] ): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] } @@ -143,7 +143,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, * partitionLeaderElectionStrategyOpt if a leader election is required. * @param partitions The partitions * @param targetState The state - * @param partitionLeaderElectionStrategyOpt The leader election strategy if a leader election is required. + * @param leaderAndIsrUpdateStrategy The leader election strategy if a leader election is required. * @return A map of failed and successful elections when targetState is OnlinePartitions. The keys are the * topic partitions and the corresponding values are either the exception that was thrown or new * leader & ISR. @@ -151,7 +151,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, override def handleStateChanges( partitions: Seq[TopicPartition], targetState: PartitionState, - partitionLeaderElectionStrategyOpt: Option[PartitionLeaderElectionStrategy] + leaderAndIsrUpdateStrategy: Option[LeaderAndIsrUpdateStrategy] ): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] = { if (partitions.nonEmpty) { try { @@ -159,7 +159,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, val result = doHandleStateChanges( partitions, targetState, - partitionLeaderElectionStrategyOpt + leaderAndIsrUpdateStrategy ) controllerBrokerRequestBatch.sendRequestsToBrokers(controllerContext.epoch) result @@ -201,6 +201,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, * --nothing other than marking the partition state as NonExistentPartition * @param partitions The partitions for which the state transition is invoked * @param targetState The end state that the partition should be moved to + * @param leaderAndIsrUpdateStrategy Strategy to use if LeaderAndIsr changes are needed * @return A map of failed and successful elections when targetState is OnlinePartitions. The keys are the * topic partitions and the corresponding values are either the exception that was thrown or new * leader & ISR. @@ -208,7 +209,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, private def doHandleStateChanges( partitions: Seq[TopicPartition], targetState: PartitionState, - partitionLeaderElectionStrategyOpt: Option[PartitionLeaderElectionStrategy] + leaderAndIsrUpdateStrategy: Option[LeaderAndIsrUpdateStrategy] ): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] = { val stateChangeLog = stateChangeLogger.withControllerEpoch(controllerContext.epoch) val traceEnabled = stateChangeLog.isTraceEnabled @@ -226,7 +227,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, Map.empty case OnlinePartition => val uninitializedPartitions = validPartitions.filter(partition => partitionState(partition) == NewPartition) - val partitionsToElectLeader = validPartitions.filter(partition => partitionState(partition) == OfflinePartition || partitionState(partition) == OnlinePartition) + val partitionsToUpdate = validPartitions.filter(partition => partitionState(partition) == OfflinePartition || partitionState(partition) == OnlinePartition) if (uninitializedPartitions.nonEmpty) { val successfulInitializations = initializeLeaderAndIsrForPartitions(uninitializedPartitions) successfulInitializations.foreach { partition => @@ -235,10 +236,10 @@ class ZkPartitionStateMachine(config: KafkaConfig, controllerContext.putPartitionState(partition, OnlinePartition) } } - if (partitionsToElectLeader.nonEmpty) { - val electionResults = electLeaderForPartitions( - partitionsToElectLeader, - partitionLeaderElectionStrategyOpt.getOrElse( + if (partitionsToUpdate.nonEmpty) { + val electionResults = updateLeaderAndIsr( + partitionsToUpdate, + leaderAndIsrUpdateStrategy.getOrElse( throw new IllegalArgumentException("Election strategy is a required field when the target state is OnlinePartition") ) ) @@ -249,7 +250,8 @@ class ZkPartitionStateMachine(config: KafkaConfig, s"Changed partition $partition from ${partitionState(partition)} to $targetState with state $leaderAndIsr" ) controllerContext.putPartitionState(partition, OnlinePartition) - case (_, Left(_)) => // Ignore; no need to update partition state on election error + case (partition, Left(e)) => + logFailedStateChange(partition, partitionState(partition), OnlinePartition, e) } electionResults @@ -321,51 +323,46 @@ class ZkPartitionStateMachine(config: KafkaConfig, /** * Repeatedly attempt to elect leaders for multiple partitions until there are no more remaining partitions to retry. * @param partitions The partitions that we're trying to elect leaders for. - * @param partitionLeaderElectionStrategy The election strategy to use. + * @param leaderAndIsrUpdateStrategy Strategy to use if LeaderAndIsr changes are needed * @return A map of failed and successful elections. The keys are the topic partitions and the corresponding values are * either the exception that was thrown or new leader & ISR. */ - private def electLeaderForPartitions( + private def updateLeaderAndIsr( partitions: Seq[TopicPartition], - partitionLeaderElectionStrategy: PartitionLeaderElectionStrategy + leaderAndIsrUpdateStrategy: LeaderAndIsrUpdateStrategy ): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] = { var remaining = partitions val finishedElections = mutable.Map.empty[TopicPartition, Either[Throwable, LeaderAndIsr]] while (remaining.nonEmpty) { - val (finished, updatesToRetry) = doElectLeaderForPartitions(remaining, partitionLeaderElectionStrategy) + val (finished, updatesToRetry) = tryUpdateLeaderAndIsr(remaining, leaderAndIsrUpdateStrategy) remaining = updatesToRetry - - finished.foreach { - case (partition, Left(e)) => - logFailedStateChange(partition, partitionState(partition), OnlinePartition, e) - case (_, Right(_)) => // Ignore; success so no need to log failed state change - } - finishedElections ++= finished - if (remaining.nonEmpty) - logger.info(s"Retrying leader election with strategy $partitionLeaderElectionStrategy for partitions $remaining") + if (remaining.nonEmpty) { + logger.info(s"Retrying LeaderAndIsr update with strategy $leaderAndIsrUpdateStrategy for " + + s"${remaining.size} partitions including ${remaining.take(5)}") + } } finishedElections.toMap } /** - * Try to elect leaders for multiple partitions. - * Electing a leader for a partition updates partition state in zookeeper. + * Try to update LeaderAndIsrs for multiple partitions. + * Altering LeaderAndIsr for a partition updates partition state in zookeeper. * * @param partitions The partitions that we're trying to elect leaders for. - * @param partitionLeaderElectionStrategy The election strategy to use. + * @param leaderAndIsrUpdateStrategy Strategy to use to update LeaderAndIsr * @return A tuple of two values: * 1. The partitions and the expected leader and isr that successfully had a leader elected. And exceptions * corresponding to failed elections that should not be retried. * 2. The partitions that we should retry due to a zookeeper BADVERSION conflict. Version conflicts can occur if * the partition leader updated partition state while the controller attempted to update partition state. */ - private def doElectLeaderForPartitions( + private def tryUpdateLeaderAndIsr( partitions: Seq[TopicPartition], - partitionLeaderElectionStrategy: PartitionLeaderElectionStrategy + leaderAndIsrUpdateStrategy: LeaderAndIsrUpdateStrategy ): (Map[TopicPartition, Either[Exception, LeaderAndIsr]], Seq[TopicPartition]) = { val getDataResponses = try { zkClient.getTopicPartitionStatesRaw(partitions) @@ -373,7 +370,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, case e: Exception => return (partitions.iterator.map(_ -> Left(e)).toMap, Seq.empty) } - val failedElections = mutable.Map.empty[TopicPartition, Either[Exception, LeaderAndIsr]] + val finished = mutable.Map.empty[TopicPartition, Either[Exception, LeaderAndIsr]] val validLeaderAndIsrs = mutable.Buffer.empty[(TopicPartition, LeaderAndIsr)] getDataResponses.foreach { getDataResponse => @@ -386,74 +383,126 @@ class ZkPartitionStateMachine(config: KafkaConfig, val failMsg = s"Aborted leader election for partition $partition since the LeaderAndIsr path was " + s"already written by another controller. This probably means that the current controller $controllerId went through " + s"a soft failure and another controller was elected with epoch ${leaderIsrAndControllerEpoch.controllerEpoch}." - failedElections.put(partition, Left(new StateChangeFailedException(failMsg))) + finished.put(partition, Left(new StateChangeFailedException(failMsg))) } else { validLeaderAndIsrs += partition -> leaderIsrAndControllerEpoch.leaderAndIsr } case None => val exception = new StateChangeFailedException(s"LeaderAndIsr information doesn't exist for partition $partition in $currState state") - failedElections.put(partition, Left(exception)) + finished.put(partition, Left(exception)) } } else if (getDataResponse.resultCode == Code.NONODE) { val exception = new StateChangeFailedException(s"LeaderAndIsr information doesn't exist for partition $partition in $currState state") - failedElections.put(partition, Left(exception)) + finished.put(partition, Left(exception)) } else { - failedElections.put(partition, Left(getDataResponse.resultException.get)) + finished.put(partition, Left(getDataResponse.resultException.get)) } } if (validLeaderAndIsrs.isEmpty) { - return (failedElections.toMap, Seq.empty) + return (finished.toMap, Seq.empty) } - val (partitionsWithoutLeaders, partitionsWithLeaders) = partitionLeaderElectionStrategy match { - case OfflinePartitionLeaderElectionStrategy(allowUnclean) => - val partitionsWithUncleanLeaderElectionState = collectUncleanLeaderElectionState( - validLeaderAndIsrs, - allowUnclean - ) - leaderForOffline( - controllerContext, - isLeaderRecoverySupported, - partitionsWithUncleanLeaderElectionState - ).partition(_.leaderAndIsr.isEmpty) - - case ReassignPartitionLeaderElectionStrategy => - leaderForReassign(controllerContext, validLeaderAndIsrs).partition(_.leaderAndIsr.isEmpty) - case PreferredReplicaPartitionLeaderElectionStrategy => - leaderForPreferredReplica(controllerContext, validLeaderAndIsrs).partition(_.leaderAndIsr.isEmpty) - case ControlledShutdownPartitionLeaderElectionStrategy => - leaderForControlledShutdown(controllerContext, validLeaderAndIsrs).partition(_.leaderAndIsr.isEmpty) + val adjustedLeaderAndIsrs = mutable.HashMap.empty[TopicPartition, LeaderAndIsr] + val neededUpdates = mutable.HashMap.empty[TopicPartition, LeaderAndIsrUpdateResult.Successful] + + processLeaderAndIsrUpdate(validLeaderAndIsrs, leaderAndIsrUpdateStrategy).foreach { + case update@LeaderAndIsrUpdateResult.Successful(topicPartition, newLeaderAndIsr, _, _, _) => + neededUpdates.put(topicPartition, update) + adjustedLeaderAndIsrs.put(topicPartition, newLeaderAndIsr) + case LeaderAndIsrUpdateResult.NotNeeded(topicPartition, currentLeaderAndIsr) => + finished.put(topicPartition, Right(currentLeaderAndIsr)) + case LeaderAndIsrUpdateResult.Failed(topicPartition, exception) => + finished.put(topicPartition, Left(exception)) } - partitionsWithoutLeaders.foreach { electionResult => - val partition = electionResult.topicPartition - val failMsg = s"Failed to elect leader for partition $partition under strategy $partitionLeaderElectionStrategy" - failedElections.put(partition, Left(new StateChangeFailedException(failMsg))) + + if (adjustedLeaderAndIsrs.isEmpty) { + return (finished.toMap, Seq.empty) } - val recipientsPerPartition = partitionsWithLeaders.map(result => result.topicPartition -> result.liveReplicas).toMap - val adjustedLeaderAndIsrs = partitionsWithLeaders.map(result => result.topicPartition -> result.leaderAndIsr.get).toMap + val UpdateLeaderAndIsrResult(finishedUpdates, updatesToRetry) = zkClient.updateLeaderAndIsr( - adjustedLeaderAndIsrs, controllerContext.epoch, controllerContext.epochZkVersion) + adjustedLeaderAndIsrs, + controllerContext.epoch, + controllerContext.epochZkVersion + ) + finishedUpdates.forKeyValue { (partition, result) => result.foreach { leaderAndIsr => val replicaAssignment = controllerContext.partitionFullReplicaAssignment(partition) val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(leaderAndIsr, controllerContext.epoch) + val leaderAndIsrUpdate = neededUpdates(partition) + val leaderAndIsrRecipients = leaderAndIsrUpdate.liveReplicas + val stopReplicaRecipients = leaderAndIsrUpdate.replicasToStop + val deleteReplicaRecipients = leaderAndIsrUpdate.replicasToDelete + controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch) - controllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers(recipientsPerPartition(partition), partition, - leaderIsrAndControllerEpoch, replicaAssignment, isNew = false) + finished.put(partition, Right(leaderAndIsr)) + + if (leaderAndIsrRecipients.nonEmpty) { + controllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers( + leaderAndIsrRecipients, + partition, + leaderIsrAndControllerEpoch, + replicaAssignment, + isNew = false + ) + } + + if (stopReplicaRecipients.nonEmpty) { + controllerBrokerRequestBatch.addStopReplicaRequestForBrokers( + stopReplicaRecipients, + partition, + deletePartition = false + ) + } + + if (deleteReplicaRecipients.nonEmpty) { + controllerBrokerRequestBatch.addStopReplicaRequestForBrokers( + deleteReplicaRecipients, + partition, + deletePartition = true + ) + } } } if (isDebugEnabled) { updatesToRetry.foreach { partition => - debug(s"Controller failed to elect leader for partition $partition. " + - s"Attempted to write state ${adjustedLeaderAndIsrs(partition)}, but failed with bad ZK version. This will be retried.") + debug(s"Controller failed to update partition $partition using $leaderAndIsrUpdateStrategy. " + + s"Attempted to write state ${adjustedLeaderAndIsrs(partition)}, but failed with bad ZK version. " + + "This will be retried.") } } - (finishedUpdates ++ failedElections, updatesToRetry) + (finished, updatesToRetry) + } + + private def processLeaderAndIsrUpdate( + currentLeaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)], + leaderAndIsrUpdateStrategy: LeaderAndIsrUpdateStrategy + ): Seq[LeaderAndIsrUpdateResult] = { + leaderAndIsrUpdateStrategy match { + case OfflineLeaderElectionStrategy(allowUnclean) => + val partitionsWithUncleanLeaderElectionState = collectUncleanLeaderElectionState( + currentLeaderAndIsrs, + allowUnclean + ) + leaderForOffline( + controllerContext, + isLeaderRecoverySupported, + partitionsWithUncleanLeaderElectionState + ) + case PreferredLeaderElectionStrategy => + leaderForPreferredReplica(controllerContext, currentLeaderAndIsrs) + case ControlledShutdownStrategy => + processControlledShutdown(controllerContext, currentLeaderAndIsrs) + case CancelReassignmentStrategy => + processReassignmentCancellation(controllerContext, currentLeaderAndIsrs) + case CompleteReassignmentStrategy => + processReassignmentCompletion(controllerContext, currentLeaderAndIsrs) + } } /* For the provided set of topic partition and partition sync state it attempts to determine if unclean @@ -466,7 +515,7 @@ class ZkPartitionStateMachine(config: KafkaConfig, * @param allowUnclean whether to allow unclean election without having to read the topic configuration * @return a sequence of three element tuple: * 1. topic partition - * 2. leader, isr and controller epoc. Some means election should be performed + * 2. leader, isr and controller epoch. Some means election should be performed * 3. allow unclean */ private def collectUncleanLeaderElectionState( @@ -555,11 +604,12 @@ object PartitionLeaderElectionAlgorithms { } } -sealed trait PartitionLeaderElectionStrategy -final case class OfflinePartitionLeaderElectionStrategy(allowUnclean: Boolean) extends PartitionLeaderElectionStrategy -final case object ReassignPartitionLeaderElectionStrategy extends PartitionLeaderElectionStrategy -final case object PreferredReplicaPartitionLeaderElectionStrategy extends PartitionLeaderElectionStrategy -final case object ControlledShutdownPartitionLeaderElectionStrategy extends PartitionLeaderElectionStrategy +sealed trait LeaderAndIsrUpdateStrategy +final case class OfflineLeaderElectionStrategy(allowUnclean: Boolean) extends LeaderAndIsrUpdateStrategy +case object PreferredLeaderElectionStrategy extends LeaderAndIsrUpdateStrategy +case object ControlledShutdownStrategy extends LeaderAndIsrUpdateStrategy +case object CancelReassignmentStrategy extends LeaderAndIsrUpdateStrategy +case object CompleteReassignmentStrategy extends LeaderAndIsrUpdateStrategy sealed trait PartitionState { def state: Byte diff --git a/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala b/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala index 5b41e66c11da8..595d1976beb9d 100644 --- a/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala +++ b/core/src/main/scala/kafka/controller/ReplicaStateMachine.scala @@ -152,6 +152,9 @@ class ZkReplicaStateMachine(config: KafkaConfig, * ReplicaDeletionSuccessful -> NonExistentReplica * -- remove the replica from the in memory partition replica assignment cache * + * NewReplica,OnlineReplica,OfflineReplica -> UnassignedReplicas + * -- Following completion or cancellation of a reassignment, remove the replica state + * * @param replicaId The replica for which the state transition is invoked * @param replicas The partitions on this replica for which the state transition is invoked * @param targetState The end state that the replica should be moved to @@ -218,22 +221,28 @@ class ZkReplicaStateMachine(config: KafkaConfig, controllerContext.putReplicaState(replica, OnlineReplica) } case OfflineReplica => - validReplicas.foreach { replica => - controllerBrokerRequestBatch.addStopReplicaRequestForBrokers(Seq(replicaId), replica.topicPartition, deletePartition = false) - } val (replicasWithLeadershipInfo, replicasWithoutLeadershipInfo) = validReplicas.partition { replica => controllerContext.partitionLeadershipInfo(replica.topicPartition).isDefined } val updatedLeaderIsrAndControllerEpochs = removeReplicasFromIsr(replicaId, replicasWithLeadershipInfo.map(_.topicPartition)) updatedLeaderIsrAndControllerEpochs.forKeyValue { (partition, leaderIsrAndControllerEpoch) => - stateLogger.info(s"Partition $partition state changed to $leaderIsrAndControllerEpoch after removing replica $replicaId from the ISR as part of transition to $OfflineReplica") + stateLogger.info(s"Partition $partition state changed to $leaderIsrAndControllerEpoch " + + s"after transitioning replica $replicaId to $OfflineReplica") if (!controllerContext.isTopicQueuedUpForDeletion(partition.topic)) { val recipients = controllerContext.partitionReplicaAssignment(partition).filterNot(_ == replicaId) controllerBrokerRequestBatch.addLeaderAndIsrRequestForBrokers(recipients, partition, leaderIsrAndControllerEpoch, - controllerContext.partitionFullReplicaAssignment(partition), isNew = false) + controllerContext.partitionFullReplicaAssignment(partition), + isNew = false + ) } + controllerBrokerRequestBatch.addStopReplicaRequestForBrokers( + Seq(replicaId), + partition, + deletePartition = false + ) + val replica = PartitionAndReplica(partition, replicaId) val currentState = controllerContext.replicaState(replica) if (traceEnabled) @@ -245,7 +254,15 @@ class ZkReplicaStateMachine(config: KafkaConfig, val currentState = controllerContext.replicaState(replica) if (traceEnabled) logSuccessfulTransition(stateLogger, replicaId, replica.topicPartition, currentState, OfflineReplica) - controllerBrokerRequestBatch.addUpdateMetadataRequestForBrokers(controllerContext.liveOrShuttingDownBrokerIds.toSeq, Set(replica.topicPartition)) + controllerBrokerRequestBatch.addUpdateMetadataRequestForBrokers( + controllerContext.liveOrShuttingDownBrokerIds.toSeq, + Set(replica.topicPartition) + ) + controllerBrokerRequestBatch.addStopReplicaRequestForBrokers( + Seq(replicaId), + replica.topicPartition, + deletePartition = false + ) controllerContext.putReplicaState(replica, OfflineReplica) } case ReplicaDeletionStarted => @@ -282,6 +299,17 @@ class ZkReplicaStateMachine(config: KafkaConfig, logSuccessfulTransition(stateLogger, replicaId, replica.topicPartition, currentState, NonExistentReplica) controllerContext.removeReplicaState(replica) } + case UnassignedReplica => + // For the case of a replica which has been removed following reassignment, + // replicas will be stopped and deleted as part of the final transition to `OnlinePartition`. + // The replica assignment also will be updated as part of the reassignment completion. + // So here we just need to remove the replica state. + validReplicas.foreach { replica => + val currentState = controllerContext.replicaState(replica) + if (traceEnabled) + logSuccessfulTransition(stateLogger, replicaId, replica.topicPartition, currentState, UnassignedReplica) + controllerContext.removeReplicaState(replica) + } } } @@ -299,14 +327,14 @@ class ZkReplicaStateMachine(config: KafkaConfig, var results = Map.empty[TopicPartition, LeaderIsrAndControllerEpoch] var remaining = partitions while (remaining.nonEmpty) { - val (finishedRemoval, removalsToRetry) = doRemoveReplicasFromIsr(replicaId, remaining) + val (finishedRemoval, removalsToRetry) = tryRemoveReplicasFromIsr(replicaId, remaining) remaining = removalsToRetry finishedRemoval.foreach { case (partition, Left(e)) => - val replica = PartitionAndReplica(partition, replicaId) - val currentState = controllerContext.replicaState(replica) - logFailedStateChange(replica, currentState, OfflineReplica, e) + val replica = PartitionAndReplica(partition, replicaId) + val currentState = controllerContext.replicaState(replica) + logFailedStateChange(replica, currentState, OfflineReplica, e) case (partition, Right(leaderIsrAndEpoch)) => results += partition -> leaderIsrAndEpoch } @@ -327,7 +355,7 @@ class ZkReplicaStateMachine(config: KafkaConfig, * 2. The partitions that we should retry due to a zookeeper BADVERSION conflict. Version conflicts can occur if * the partition leader updated partition state while the controller attempted to update partition state. */ - private def doRemoveReplicasFromIsr( + private def tryRemoveReplicasFromIsr( replicaId: Int, partitions: Seq[TopicPartition] ): (Map[TopicPartition, Either[Exception, LeaderIsrAndControllerEpoch]], Seq[TopicPartition]) = { @@ -489,3 +517,8 @@ case object NonExistentReplica extends ReplicaState { val state: Byte = 7 val validPreviousStates: Set[ReplicaState] = Set(ReplicaDeletionSuccessful) } + +case object UnassignedReplica extends ReplicaState { + val state: Byte = 8 + val validPreviousStates: Set[ReplicaState] = Set(NewReplica, OnlineReplica, OfflineReplica) +} diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala index 64cf88d4eef3a..390bd4a1c7e95 100755 --- a/core/src/main/scala/kafka/server/KafkaServer.scala +++ b/core/src/main/scala/kafka/server/KafkaServer.scala @@ -584,7 +584,6 @@ class KafkaServer( var shutdownSucceeded: Boolean = false try { - var remainingRetries = retries var prevController: Node = null var ioException = false diff --git a/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala b/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala index 29b4c82740e2a..be28740fa3df0 100644 --- a/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala @@ -24,6 +24,7 @@ import kafka.utils.{TestInfoUtils, TestUtils} import org.apache.kafka.clients.admin._ import org.apache.kafka.clients.producer.ProducerRecord import org.apache.kafka.common.config.ConfigResource +import org.apache.kafka.common.errors.InvalidReplicaAssignmentException import org.apache.kafka.common.utils.Utils import org.apache.kafka.common.{TopicPartition, TopicPartitionReplica} import org.apache.kafka.server.common.MetadataVersion.IBP_2_7_IV1 @@ -33,7 +34,7 @@ import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.ValueSource import java.io.Closeable -import java.util.{Collections, HashMap, List} +import java.util.{Collections, HashMap, List, Optional} import scala.collection.{Map, Seq, mutable} import scala.jdk.CollectionConverters._ @@ -108,12 +109,13 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { """{"topic":"bar","partition":0,"replicas":[3,2,0],"log_dirs":["any","any","any"]}""" + """]}""" + val foo0 = new TopicPartition("foo", 0) + val bar0 = new TopicPartition("bar", 0) + // Check that the assignment has not yet been started yet. val initialAssignment = Map( - new TopicPartition("foo", 0) -> - PartitionReassignmentState(Seq(0, 1, 2), Seq(0, 1, 3), true), - new TopicPartition("bar", 0) -> - PartitionReassignmentState(Seq(3, 2, 1), Seq(3, 2, 0), true) + foo0 -> PartitionReassignmentState(Seq(0, 1, 2), Seq(0, 1, 3), true), + bar0 -> PartitionReassignmentState(Seq(3, 2, 1), Seq(3, 2, 0), true) ) waitForVerifyAssignment(cluster.adminClient, assignment, false, VerifyAssignmentResult(initialAssignment)) @@ -122,10 +124,8 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { runExecuteAssignment(cluster.adminClient, false, assignment, -1L, -1L) assertEquals(unthrottledBrokerConfigs, describeBrokerLevelThrottles(unthrottledBrokerConfigs.keySet.toSeq)) val finalAssignment = Map( - new TopicPartition("foo", 0) -> - PartitionReassignmentState(Seq(0, 1, 3), Seq(0, 1, 3), true), - new TopicPartition("bar", 0) -> - PartitionReassignmentState(Seq(3, 2, 0), Seq(3, 2, 0), true) + foo0 -> PartitionReassignmentState(Seq(0, 1, 3), Seq(0, 1, 3), true), + bar0 -> PartitionReassignmentState(Seq(3, 2, 0), Seq(3, 2, 0), true) ) val verifyAssignmentResult = runVerifyAssignment(cluster.adminClient, assignment, false) @@ -137,6 +137,10 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { assertEquals(unthrottledBrokerConfigs, describeBrokerLevelThrottles(unthrottledBrokerConfigs.keySet.toSeq)) + + // Verify that partitions are removed from brokers no longer assigned + verifyReplicaDeleted(topicPartition = foo0, replicaId = 2) + verifyReplicaDeleted(topicPartition = bar0, replicaId = 1) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @@ -296,10 +300,13 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @ValueSource(strings = Array("zk", "kraft")) def testCancellation(quorum: String): Unit = { + val foo0 = new TopicPartition("foo", 0) + val baz1 = new TopicPartition("baz", 1) + cluster = new ReassignPartitionsTestCluster() cluster.setup() - cluster.produceMessages("foo", 0, 200) - cluster.produceMessages("baz", 1, 200) + cluster.produceMessages(foo0.topic, foo0.partition, 200) + cluster.produceMessages(baz1.topic, baz1.partition, 200) val assignment = """{"version":1,"partitions":""" + """[{"topic":"foo","partition":0,"replicas":[0,1,3],"log_dirs":["any","any","any"]},""" + """{"topic":"baz","partition":1,"replicas":[0,2,3],"log_dirs":["any","any","any"]}""" + @@ -314,14 +321,11 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { // from completing before this runs. waitForVerifyAssignment(cluster.adminClient, assignment, true, VerifyAssignmentResult(Map( - new TopicPartition("foo", 0) -> PartitionReassignmentState(Seq(0, 1, 3, 2), Seq(0, 1, 3), false), - new TopicPartition("baz", 1) -> PartitionReassignmentState(Seq(0, 2, 3, 1), Seq(0, 2, 3), false)), + foo0 -> PartitionReassignmentState(Seq(0, 1, 3, 2), Seq(0, 1, 3), false), + baz1 -> PartitionReassignmentState(Seq(0, 2, 3, 1), Seq(0, 2, 3), false)), true, Map(), false)) // Cancel the reassignment. - assertEquals((Set( - new TopicPartition("foo", 0), - new TopicPartition("baz", 1) - ), Set()), runCancelAssignment(cluster.adminClient, assignment, true)) + assertEquals((Set(foo0, baz1), Set()), runCancelAssignment(cluster.adminClient, assignment, true)) // Broker throttles are still active because we passed --preserve-throttles waitForInterBrokerThrottle(Set(0, 1, 2, 3), interBrokerThrottle) // Cancelling the reassignment again should reveal nothing to cancel. @@ -330,6 +334,118 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { waitForBrokerLevelThrottles(unthrottledBrokerConfigs) // Verify that there are no ongoing reassignments. assertFalse(runVerifyAssignment(cluster.adminClient, assignment, false).partsOngoing) + // Verify that the partition is removed from cancelled replicas + verifyReplicaDeleted(topicPartition = foo0, replicaId = 3) + verifyReplicaDeleted(topicPartition = baz1, replicaId = 3) + } + + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk", "kraft")) + def testCancellationWithAddingReplicaInIsr(quorum: String): Unit = { + val foo0 = new TopicPartition("foo", 0) + + cluster = new ReassignPartitionsTestCluster() + cluster.setup() + cluster.produceMessages(foo0.topic, foo0.partition, 200) + + // The reassignment will bring replicas 3 and 4 into the replica set and remove 1 and 2. + val assignment = """{"version":1,"partitions":""" + + """[{"topic":"foo","partition":0,"replicas":[0,3,4],"log_dirs":["any","any","any"]}""" + + """]}""" + + // We will throttle replica 4 so that only replica 3 joins the ISR + TestUtils.setReplicationThrottleForPartitions( + cluster.adminClient, + brokerIds = Seq(4), + partitions = Set(foo0), + throttleBytes = 1 + ) + + // Execute the assignment and wait for replica 3 (only) to join the ISR + runExecuteAssignment( + cluster.adminClient, + additional = false, + reassignmentJson = assignment + ) + TestUtils.waitUntilTrue( + () => TestUtils.currentIsr(cluster.adminClient, foo0) == Set(0, 1, 2, 3), + msg = "Timed out while waiting for replica 3 to join the ISR" + ) + + // Now cancel the assignment and verify that the partition is removed from cancelled replicas + assertEquals((Set(foo0), Set()), runCancelAssignment(cluster.adminClient, assignment, preserveThrottles = true)) + verifyReplicaDeleted(topicPartition = foo0, replicaId = 3) + verifyReplicaDeleted(topicPartition = foo0, replicaId = 4) + } + + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk", "kraft")) + def testCancellationNotAllowedIfOnlyAddingReplicasInIsr(quorum: String): Unit = { + val boo0 = new TopicPartition("boo", 0) + + cluster = new ReassignPartitionsTestCluster() + cluster.setup() + cluster.produceMessages(boo0.topic, boo0.partition, 200) + + // The reassignment will add replicas 1 and 2 + val assignment = """{"version":1,"partitions":""" + + """[{"topic":"boo","partition":0,"replicas":[0,1,2],"log_dirs":["any","any","any"]}""" + + """]}""" + + // We will throttle replica 2 so that only replica 1 joins the ISR + TestUtils.setReplicationThrottleForPartitions( + cluster.adminClient, + brokerIds = Seq(2), + partitions = Set(boo0), + throttleBytes = 1 + ) + + // Execute the assignment and wait for replica 1 (only) to join the ISR + runExecuteAssignment( + cluster.adminClient, + additional = false, + reassignmentJson = assignment + ) + TestUtils.waitUntilTrue( + () => TestUtils.currentIsr(cluster.adminClient, boo0) == Set(0, 1), + msg = "Timed out while waiting for replica 3 to join the ISR" + ) + + // Now we shutdown broker 0 and wait for 1 to become the leader + cluster.servers(0).shutdown() + cluster.servers(0).awaitShutdown() + + TestUtils.awaitLeaderChange( + cluster.servers, + boo0, + oldLeader = 0 + ) + + TestUtils.waitUntilTrue( + () => TestUtils.currentIsr(cluster.adminClient, boo0) == Set(1), + msg = "Timed out while waiting for replica 3 to join the ISR" + ) + + // Reassignment cancellation should not be allowed since it would leave no remaining + // replicas in the ISR. + val future = cluster.adminClient.alterPartitionReassignments( + Map(boo0 -> Optional.empty[NewPartitionReassignment]()).asJava + ).all() + TestUtils.assertFutureExceptionTypeEquals(future, classOf[InvalidReplicaAssignmentException]) + } + + private def verifyReplicaDeleted( + topicPartition: TopicPartition, + replicaId: Int + ): Unit = { + def isReplicaStoppedAndDeleted(): Boolean = { + val server = cluster.servers(replicaId) + val partition = server.replicaManager.getPartition(topicPartition) + val log = server.logManager.getLog(topicPartition) + partition == HostedPartition.None && log.isEmpty + } + TestUtils.waitUntilTrue(isReplicaStoppedAndDeleted, + msg = s"Timed out waiting for replica $replicaId of $topicPartition to be deleted") } private def waitForLogDirThrottle(throttledBrokers: Set[Int], logDirThrottle: Long): Unit = { @@ -541,8 +657,8 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { private def runExecuteAssignment(adminClient: Admin, additional: Boolean, reassignmentJson: String, - interBrokerThrottle: Long, - replicaAlterLogDirsThrottle: Long) = { + interBrokerThrottle: Long = -1, + replicaAlterLogDirsThrottle: Long = -1) = { println(s"==> executeAssignment(adminClient, additional=${additional}, " + s"reassignmentJson=${reassignmentJson}, " + s"interBrokerThrottle=${interBrokerThrottle}, " + @@ -591,7 +707,8 @@ class ReassignPartitionsIntegrationTest extends QuorumTestHarness { val topics = Map( "foo" -> Seq(Seq(0, 1, 2), Seq(1, 2, 3)), "bar" -> Seq(Seq(3, 2, 1)), - "baz" -> Seq(Seq(1, 0, 2), Seq(2, 0, 1), Seq(0, 2, 1)) + "baz" -> Seq(Seq(1, 0, 2), Seq(2, 0, 1), Seq(0, 2, 1)), + "boo" -> Seq(Seq(0)) ) val brokerConfigs = brokers.map { diff --git a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala index d563bcb0c82c3..fcd9c484f9f32 100644 --- a/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala @@ -1535,7 +1535,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest { s"Unexpected message: ${exception.getMessage}") } else { assertTrue(exception.getMessage.contains( - s"Failed to elect leader for partition $topicPartition under strategy PreferredReplicaPartitionLeaderElectionStrategy"), + s"Failed to elect preferred leader"), s"Unexpected message: ${exception.getMessage}") } } diff --git a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala index d49502dd628a2..bc4aa522f2578 100644 --- a/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala +++ b/core/src/test/scala/unit/kafka/controller/ControllerIntegrationTest.scala @@ -325,7 +325,8 @@ class ControllerIntegrationTest extends QuorumTestHarness { val reassignment = Map(tp -> ReplicaAssignment(Seq(otherBrokerId), List(), List())) TestUtils.createTopic(zkClient, tp.topic, partitionReplicaAssignment = assignment, servers = servers) zkClient.createPartitionReassignment(reassignment.map { case (k, v) => k -> v.replicas }) - waitForPartitionState(tp, firstControllerEpoch, otherBrokerId, LeaderAndIsr.InitialLeaderEpoch + 3, + // One leader epoch bump needed to add the new replica, one leader epoch bump to remove the old one + waitForPartitionState(tp, firstControllerEpoch, otherBrokerId, LeaderAndIsr.InitialLeaderEpoch + 2, "failed to get expected partition state after partition reassignment") TestUtils.waitUntilTrue(() => zkClient.getFullReplicaAssignmentForTopics(Set(tp.topic)) == reassignment, "failed to get updated partition assignment on topic znode after partition reassignment") @@ -364,7 +365,7 @@ class ControllerIntegrationTest extends QuorumTestHarness { val reassignment = Map(tp -> ReplicaAssignment(Seq(otherBrokerId), List(), List())) TestUtils.createTopic(zkClient, tp.topic, partitionReplicaAssignment = assignment, servers = servers) zkClient.createPartitionReassignment(reassignment.map { case (k, v) => k -> v.replicas }) - waitForPartitionState(tp, firstControllerEpoch, otherBrokerId, LeaderAndIsr.InitialLeaderEpoch + 3, + waitForPartitionState(tp, firstControllerEpoch, otherBrokerId, LeaderAndIsr.InitialLeaderEpoch + 2, "with an offline log directory on the target broker, the partition reassignment stalls") TestUtils.waitUntilTrue(() => zkClient.getFullReplicaAssignmentForTopics(Set(tp.topic)) == reassignment, "failed to get updated partition assignment on topic znode after partition reassignment") @@ -410,7 +411,7 @@ class ControllerIntegrationTest extends QuorumTestHarness { waitForPartitionState(tp, firstControllerEpoch, controllerId, LeaderAndIsr.InitialLeaderEpoch + 1, "failed to get expected partition state during partition reassignment with offline replica") servers(otherBrokerId).startup() - waitForPartitionState(tp, firstControllerEpoch, otherBrokerId, LeaderAndIsr.InitialLeaderEpoch + 4, + waitForPartitionState(tp, firstControllerEpoch, otherBrokerId, LeaderAndIsr.InitialLeaderEpoch + 3, "failed to get expected partition state after partition reassignment") TestUtils.waitUntilTrue(() => zkClient.getFullReplicaAssignmentForTopics(Set(tp.topic)) == reassignment, "failed to get updated partition assignment on topic znode after partition reassignment") @@ -618,13 +619,12 @@ class ControllerIntegrationTest extends QuorumTestHarness { @Test def testControllerMoveOnPartitionReassignment(): Unit = { - servers = makeServers(1) + servers = makeServers(numConfigs = 2) TestUtils.waitUntilControllerElected(zkClient) val tp = new TopicPartition("t", 0) val assignment = Map(tp.partition -> Seq(0)) TestUtils.createTopic(zkClient, tp.topic(), assignment, servers) - - val reassignment = Map(tp -> Seq(0)) + val reassignment = Map(tp -> Seq(1)) testControllerMove(() => zkClient.createPartitionReassignment(reassignment)) } @@ -1447,9 +1447,10 @@ class ControllerIntegrationTest extends QuorumTestHarness { @Test def testTopicIdUpgradeAfterReassigningPartitions(): Unit = { + // FIXME: This test doesn't work because the reassignment is a no-op val tp = new TopicPartition("t", 0) - val reassignment = Map(tp -> Some(Seq(0))) val adminZkClient = new AdminZkClient(zkClient) + val reassignment = Map(tp -> Some(Seq(0))) // start server with old IBP servers = makeServers(1, interBrokerProtocolVersion = Some(IBP_2_7_IV0)) @@ -1564,6 +1565,7 @@ class ControllerIntegrationTest extends QuorumTestHarness { replicas: Set[Int], leaderEpoch: Int): Unit = { otherBroker.shutdown() otherBroker.awaitShutdown() + waitForPartitionState(tp, firstControllerEpoch, controllerId, leaderEpoch + 1, "failed to get expected partition state upon broker shutdown") otherBroker.startup() diff --git a/core/src/test/scala/unit/kafka/controller/ElectionTest.scala b/core/src/test/scala/unit/kafka/controller/ElectionTest.scala new file mode 100644 index 0000000000000..db564e6277ad1 --- /dev/null +++ b/core/src/test/scala/unit/kafka/controller/ElectionTest.scala @@ -0,0 +1,828 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package kafka.controller + +import kafka.api.LeaderAndIsr +import org.apache.kafka.common.TopicPartition +import org.apache.kafka.metadata.LeaderRecoveryState +import org.junit.jupiter.api.Assertions.{assertEquals, assertThrows, fail} +import org.junit.jupiter.api.Test +import org.mockito.Mockito + +import scala.collection.mutable + +class ElectionTest { + private val topicPartition = new TopicPartition("foo", 15) + + @Test + def testControlledShutdownIfAnotherLiveReplicaIsInIsr(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 5, + isr = List(1, 2), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(2, 3), + shuttingDownBrokerIds = Seq(1) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 2, + leaderEpoch = 6, + isr = List(2), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(2, 3), + replicasToStop = Seq(1) + ), result) + } + + @Test + def testControlledShutdownIfShuttingDownBrokerIsOnlyMemberOfIsr(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 5, + isr = List(1), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertFailedUpdate(controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(2, 3), + shuttingDownBrokerIds = Seq(1) + )) + } + + @Test + def testControlledShutdownIfNoActiveLeader(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = LeaderAndIsr.NoLeader, + leaderEpoch = 5, + isr = List(2), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3), + shuttingDownBrokerIds = Seq(1) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = LeaderAndIsr.NoLeader, + leaderEpoch = 6, + isr = List(2), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(3), + replicasToStop = Seq(1) + ), result) + } + + @Test + def testControlledShutdownIfShuttingDownBrokerNotInIsr(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 2, + leaderEpoch = 5, + isr = List(2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(2, 3), + shuttingDownBrokerIds = Seq(1) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 2, + leaderEpoch = 6, + isr = List(2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(2, 3), + replicasToStop = Seq(1) + ), result) + } + + @Test + def testControlledShutdownMultipleBrokersInIsr(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 2, + leaderEpoch = 5, + isr = List(1, 2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3), + shuttingDownBrokerIds = Seq(1, 2) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 3, + leaderEpoch = 6, + isr = List(3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(3), + replicasToStop = Seq(1, 2) + ), result) + } + + @Test + def testControlledShutdownMultipleBrokersOnlyMembersInIsr(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 2, + leaderEpoch = 5, + isr = List(1, 2), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3), + shuttingDownBrokerIds = Seq(1, 2) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 2, + leaderEpoch = 6, + isr = List(2), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(3, 2), + replicasToStop = Seq(1) + ), result) + } + + @Test + def testControlledShutdownMultipleBrokersAndNoCurrentLeader(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = LeaderAndIsr.NoLeader, + leaderEpoch = 5, + isr = List(3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(), + shuttingDownBrokerIds = Seq(1, 2) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = LeaderAndIsr.NoLeader, + leaderEpoch = 6, + isr = List(3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(), + replicasToStop = Seq(1, 2) + ), result) + } + + @Test + def testControlledShutdownMultipleBrokersNotInIsr(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 3, + leaderEpoch = 5, + isr = List(3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3), + shuttingDownBrokerIds = Seq(1, 2) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 3, + leaderEpoch = 6, + isr = List(3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(3), + replicasToStop = Seq(1, 2) + ), result) + } + + @Test + def testControlledShutdownMultipleBrokersOneInIsrAndOneNot(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 3, + leaderEpoch = 5, + isr = List(2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3), + shuttingDownBrokerIds = Seq(1, 2) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 3, + leaderEpoch = 6, + isr = List(3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(3), + replicasToStop = Seq(1, 2) + ), result) + } + + @Test + def testControlledShutdownSingleReplica(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + val initialLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 5, + isr = List(1), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertFailedUpdate(controlledShutdown( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(), + shuttingDownBrokerIds = Seq(1) + )) + } + + @Test + def testCancelReassignmentNoAddingReplicasInIsr(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 5, + isr = List(1, 2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 6, + isr = List(1, 2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(1, 2, 3), + replicasToDelete = Seq(4, 5) + ), result) + } + + @Test + def testCancelReassignmentWithAddingReplicasInIsr(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 5, + isr = List(1, 2, 3, 5), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 6, + isr = List(1, 2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(1, 2, 3), + replicasToDelete = Seq(4, 5) + ), result) + } + + @Test + def testCancelReassignmentWithLeaderInTargetReplicas(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 4, + leaderEpoch = 5, + isr = List(1, 2, 3, 4), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 6, + isr = List(1, 2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(1, 2, 3), + replicasToDelete = Seq(4, 5) + ), result) + } + + @Test + def testCancelReassignmentWithPreferredLeaderOffline(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 4, + leaderEpoch = 5, + isr = List(3, 4), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3, 4, 5) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 3, + leaderEpoch = 6, + isr = List(3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(3), + replicasToDelete = Seq(4, 5) + ), result) + } + + @Test + def testCancelReassignmentWithNoOriginalReplicasInIsr(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 4, + leaderEpoch = 5, + isr = List(4), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertFailedUpdate(cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3, 4, 5) + )) + } + + @Test + def testCancelReassignmentWithNoLiveOriginalReplicasInIsr(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2), + addingReplicas = Seq(2), + removingReplicas = Seq(1) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = LeaderAndIsr.NoLeader, + leaderEpoch = 5, + isr = List(1), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertFailedUpdate(cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(2) + )) + } + + @Test + def testCancelReassignmentWithNoReassignmentInProgress(): Unit = { + val assignment = ReplicaAssignment(Seq(1, 2, 3)) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 5, + isr = List(1, 2, 3), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertThrows(classOf[IllegalStateException], () => cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(3, 4, 5) + )) + } + + @Test + def testCancelReassignmentWithNoAddingReplicas(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(), + removingReplicas = Seq(4, 5) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 4, + leaderEpoch = 5, + isr = List(3, 4), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = cancelReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + ) + + assertEquals(LeaderAndIsrUpdateResult.NotNeeded( + topicPartition, + initialLeaderAndIsr + ), result) + } + + @Test + def testCompleteReassignmentCurrentLeaderNotInTargetReplicas(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 2, + leaderEpoch = 5, + isr = List(1, 2, 3, 4, 5), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = completeReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 6, + isr = List(1, 4, 5), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(1, 4, 5), + replicasToDelete = Seq(2, 3) + ), result) + } + + @Test + def testCompleteReassignmentCurrentLeaderInTargetReplicas(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 4, + leaderEpoch = 5, + isr = List(1, 2, 3, 4, 5), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = completeReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + ) + + val expectedLeaderAndIsr = LeaderAndIsr( + leader = 4, + leaderEpoch = 6, + isr = List(1, 4, 5), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertEquals(LeaderAndIsrUpdateResult.Successful( + topicPartition, + expectedLeaderAndIsr, + liveReplicas = Seq(1, 4, 5), + replicasToDelete = Seq(2, 3) + ), result) + } + + @Test + def testCompleteReassignmentWithNoReplicasToRemove(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq() + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = 1, + leaderEpoch = 5, + isr = List(1, 2, 3, 4, 5), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + val result = completeReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + ) + + assertEquals(LeaderAndIsrUpdateResult.NotNeeded( + topicPartition, + initialLeaderAndIsr + ), result) + } + + @Test + def testCompleteReassignmentWithSomeTargetReplicasOutOfIsr(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2, 3, 4, 5), + addingReplicas = Seq(4, 5), + removingReplicas = Seq(2, 3) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = LeaderAndIsr.NoLeader, + leaderEpoch = 1, + isr = List(1, 2, 3, 4), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertFailedUpdate(completeReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1, 2, 3, 4, 5) + )) + } + + @Test + def testCompleteReassignmentWithNoLiveTargetReplicasInIsr(): Unit = { + val assignment = ReplicaAssignment( + replicas = Seq(1, 2), + addingReplicas = Seq(2), + removingReplicas = Seq(1) + ) + + val initialLeaderAndIsr = LeaderAndIsr( + leader = LeaderAndIsr.NoLeader, + leaderEpoch = 5, + isr = List(2), + LeaderRecoveryState.RECOVERED, + partitionEpoch = 79 + ) + + assertFailedUpdate(completeReassignment( + initialLeaderAndIsr, + assignment, + liveBrokerIds = Seq(1) + )) + } + + private def assertFailedUpdate( + result: LeaderAndIsrUpdateResult + ): Unit = { + result match { + case LeaderAndIsrUpdateResult.Failed(topicPartition, _) => + assertEquals(this.topicPartition, topicPartition) + case _ => + fail(s"Unexpected result: $result") + } + } + + private def cancelReassignment( + initialLeaderAndIsr: LeaderAndIsr, + assignment: ReplicaAssignment, + liveBrokerIds: Seq[Int], + shuttingDownBrokerIds: Seq[Int] = Seq() + ): LeaderAndIsrUpdateResult = { + val controllerContext = mockedControllerContext( + topicPartition, + assignment, + liveBrokerIds, + shuttingDownBrokerIds + ) + + val results = Election.processReassignmentCancellation( + controllerContext = controllerContext, + Seq(topicPartition -> initialLeaderAndIsr) + ) + assertEquals(1, results.size) + assertEquals(topicPartition, results.head.topicPartition) + results.head + } + + private def completeReassignment( + initialLeaderAndIsr: LeaderAndIsr, + assignment: ReplicaAssignment, + liveBrokerIds: Seq[Int], + shuttingDownBrokerIds: Seq[Int] = Seq() + ): LeaderAndIsrUpdateResult = { + val controllerContext = mockedControllerContext( + topicPartition, + assignment, + liveBrokerIds, + shuttingDownBrokerIds + ) + + val results = Election.processReassignmentCompletion( + controllerContext = controllerContext, + Seq(topicPartition -> initialLeaderAndIsr) + ) + assertEquals(1, results.size) + assertEquals(topicPartition, results.head.topicPartition) + results.head + } + private def controlledShutdown( + initialLeaderAndIsr: LeaderAndIsr, + assignment: ReplicaAssignment, + liveBrokerIds: Seq[Int], + shuttingDownBrokerIds: Seq[Int] + + ): LeaderAndIsrUpdateResult = { + val controllerContext = mockedControllerContext( + topicPartition, + assignment, + liveBrokerIds, + shuttingDownBrokerIds + ) + + val results = Election.processControlledShutdown( + controllerContext = controllerContext, + Seq(topicPartition -> initialLeaderAndIsr) + ) + assertEquals(1, results.size) + assertEquals(topicPartition, results.head.topicPartition) + results.head + } + + private def mockedControllerContext( + topicPartition: TopicPartition, + assignment: ReplicaAssignment, + liveBrokerIds: Seq[Int], + shuttingDownBrokerIds: Seq[Int] + ): ControllerContext = { + val controllerContext = Mockito.mock(classOf[ControllerContext]) + Mockito.when(controllerContext.partitionReplicaAssignment(topicPartition)) + .thenReturn(assignment.replicas) + Mockito.when(controllerContext.partitionFullReplicaAssignment(topicPartition)) + .thenReturn(assignment) + + Mockito.when(controllerContext.shuttingDownBrokerIds) + .thenReturn(mutable.Set(shuttingDownBrokerIds: _*)) + + assignment.replicas.foreach { replicaId => + Mockito.when(controllerContext.isReplicaOnline(replicaId, topicPartition, includeShuttingDownBrokers = false)) + .thenReturn(liveBrokerIds.contains(replicaId)) + } + + controllerContext + } + +} diff --git a/core/src/test/scala/unit/kafka/controller/MockPartitionStateMachine.scala b/core/src/test/scala/unit/kafka/controller/MockPartitionStateMachine.scala index 9bc6e3cd6344d..48f7a14441b33 100644 --- a/core/src/test/scala/unit/kafka/controller/MockPartitionStateMachine.scala +++ b/core/src/test/scala/unit/kafka/controller/MockPartitionStateMachine.scala @@ -42,7 +42,7 @@ class MockPartitionStateMachine( override def handleStateChanges( partitions: Seq[TopicPartition], targetState: PartitionState, - leaderElectionStrategy: Option[PartitionLeaderElectionStrategy] + leaderElectionStrategy: Option[LeaderAndIsrUpdateStrategy] ): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] = { stateChangesByTargetState(targetState) = stateChangesByTargetState(targetState) + 1 @@ -81,7 +81,7 @@ class MockPartitionStateMachine( private def doLeaderElections( partitions: Seq[TopicPartition], - leaderElectionStrategy: PartitionLeaderElectionStrategy + leaderElectionStrategy: LeaderAndIsrUpdateStrategy ): Map[TopicPartition, Either[Throwable, LeaderAndIsr]] = { val failedElections = mutable.Map.empty[TopicPartition, Either[Throwable, LeaderAndIsr]] val validLeaderAndIsrs = mutable.Buffer.empty[(TopicPartition, LeaderAndIsr)] @@ -99,7 +99,7 @@ class MockPartitionStateMachine( } val electionResults = leaderElectionStrategy match { - case OfflinePartitionLeaderElectionStrategy(isUnclean) => + case OfflineLeaderElectionStrategy(isUnclean) => val partitionsWithUncleanLeaderElectionState = validLeaderAndIsrs.map { case (partition, leaderAndIsr) => (partition, Some(leaderAndIsr), isUnclean || uncleanLeaderElectionEnabled) } @@ -108,27 +108,25 @@ class MockPartitionStateMachine( isLeaderRecoverySupported, partitionsWithUncleanLeaderElectionState ) - case ReassignPartitionLeaderElectionStrategy => - leaderForReassign(controllerContext, validLeaderAndIsrs) - case PreferredReplicaPartitionLeaderElectionStrategy => + case PreferredLeaderElectionStrategy => leaderForPreferredReplica(controllerContext, validLeaderAndIsrs) - case ControlledShutdownPartitionLeaderElectionStrategy => - leaderForControlledShutdown(controllerContext, validLeaderAndIsrs) + case ControlledShutdownStrategy => + processControlledShutdown(controllerContext, validLeaderAndIsrs) + case CancelReassignmentStrategy => + processReassignmentCancellation(controllerContext, validLeaderAndIsrs) + case CompleteReassignmentStrategy => + processReassignmentCompletion(controllerContext, validLeaderAndIsrs) } - val results: Map[TopicPartition, Either[Exception, LeaderAndIsr]] = electionResults.map { electionResult => - val partition = electionResult.topicPartition - val value = electionResult.leaderAndIsr match { - case None => - val failMsg = s"Failed to elect leader for partition $partition under strategy $leaderElectionStrategy" - Left(new StateChangeFailedException(failMsg)) - case Some(leaderAndIsr) => - val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(leaderAndIsr, controllerContext.epoch) - controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch) - Right(leaderAndIsr) - } - - partition -> value + val results: Map[TopicPartition, Either[Exception, LeaderAndIsr]] = electionResults.map { + case LeaderAndIsrUpdateResult.Failed(partition, exception) => + partition -> Left(exception) + case LeaderAndIsrUpdateResult.NotNeeded(partition, currentLeaderAndIsr) => + partition -> Right(currentLeaderAndIsr) + case LeaderAndIsrUpdateResult.Successful(partition, newLeaderAndIsr, _, _, _) => + val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(newLeaderAndIsr, controllerContext.epoch) + controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch) + partition -> Right(newLeaderAndIsr) }.toMap results ++ failedElections diff --git a/core/src/test/scala/unit/kafka/controller/PartitionStateMachineTest.scala b/core/src/test/scala/unit/kafka/controller/PartitionStateMachineTest.scala index 9f11d42e697c6..e6535980884cc 100644 --- a/core/src/test/scala/unit/kafka/controller/PartitionStateMachineTest.scala +++ b/core/src/test/scala/unit/kafka/controller/PartitionStateMachineTest.scala @@ -17,6 +17,7 @@ package kafka.controller import kafka.api.LeaderAndIsr +import kafka.common.StateChangeFailedException import kafka.log.LogConfig import kafka.server.KafkaConfig import kafka.utils.TestUtils @@ -71,7 +72,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(false)) + Option(OfflineLeaderElectionStrategy(false)) ) assertEquals(NonExistentPartition, partitionState(partition)) } @@ -93,7 +94,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(false)) + Option(OfflineLeaderElectionStrategy(false)) ) verify(mockControllerBrokerRequestBatch).newBatch() verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(Seq(brokerId), @@ -114,7 +115,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(false)) + Option(OfflineLeaderElectionStrategy(false)) ) verify(mockControllerBrokerRequestBatch).newBatch() verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) @@ -133,7 +134,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(false)) + Option(OfflineLeaderElectionStrategy(false)) ) verify(mockControllerBrokerRequestBatch).newBatch() verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) @@ -156,11 +157,14 @@ class PartitionStateMachineTest { } @Test - def testOnlinePartitionToOnlineTransition(): Unit = { - controllerContext.setLiveBrokers(Map(TestUtils.createBrokerAndEpoch(brokerId, "host", 0))) - controllerContext.updatePartitionFullReplicaAssignment(partition, ReplicaAssignment(Seq(brokerId))) + def testOnlinePartitionToOnlineTransitionWithPreferredLeaderElectionNeeded(): Unit = { + val otherBrokerId = brokerId + 1 + controllerContext.setLiveBrokers(Map( + TestUtils.createBrokerAndEpoch(brokerId, "host", 0), + TestUtils.createBrokerAndEpoch(otherBrokerId, "host", 0))) + controllerContext.updatePartitionFullReplicaAssignment(partition, ReplicaAssignment(Seq(brokerId, otherBrokerId))) controllerContext.putPartitionState(partition, OnlinePartition) - val leaderAndIsr = LeaderAndIsr(brokerId, List(brokerId)) + val leaderAndIsr = LeaderAndIsr(otherBrokerId, List(brokerId, otherBrokerId)) val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(leaderAndIsr, controllerEpoch) controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch) @@ -174,10 +178,11 @@ class PartitionStateMachineTest { when(mockZkClient.updateLeaderAndIsr(Map(partition -> leaderAndIsrAfterElection), controllerEpoch, controllerContext.epochZkVersion)) .thenReturn(UpdateLeaderAndIsrResult(Map(partition -> Right(updatedLeaderAndIsr)), Seq.empty)) - partitionStateMachine.handleStateChanges(partitions, OnlinePartition, Option(PreferredReplicaPartitionLeaderElectionStrategy)) + partitionStateMachine.handleStateChanges(partitions, OnlinePartition, Option(PreferredLeaderElectionStrategy)) verify(mockControllerBrokerRequestBatch).newBatch() - verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(Seq(brokerId), - partition, LeaderIsrAndControllerEpoch(updatedLeaderAndIsr, controllerEpoch), replicaAssignment(Seq(brokerId)), isNew = false) + verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(Seq(brokerId, otherBrokerId), + partition, LeaderIsrAndControllerEpoch(updatedLeaderAndIsr, controllerEpoch), + replicaAssignment(Seq(brokerId, otherBrokerId)), isNew = false) verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) verify(mockZkClient).getTopicPartitionStatesRaw(any()) verify(mockZkClient).updateLeaderAndIsr(any(), anyInt(), anyInt()) @@ -185,16 +190,12 @@ class PartitionStateMachineTest { } @Test - def testOnlinePartitionToOnlineTransitionForControlledShutdown(): Unit = { + def testOnlinePartitionToOnlineTransitionWithPreferredLeaderElectionNotNeeded(): Unit = { val otherBrokerId = brokerId + 1 controllerContext.setLiveBrokers(Map( TestUtils.createBrokerAndEpoch(brokerId, "host", 0), TestUtils.createBrokerAndEpoch(otherBrokerId, "host", 0))) - controllerContext.shuttingDownBrokerIds.add(brokerId) - controllerContext.updatePartitionFullReplicaAssignment( - partition, - ReplicaAssignment(Seq(brokerId, otherBrokerId)) - ) + controllerContext.updatePartitionFullReplicaAssignment(partition, ReplicaAssignment(Seq(brokerId, otherBrokerId))) controllerContext.putPartitionState(partition, OnlinePartition) val leaderAndIsr = LeaderAndIsr(brokerId, List(brokerId, otherBrokerId)) val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(leaderAndIsr, controllerEpoch) @@ -204,17 +205,88 @@ class PartitionStateMachineTest { when(mockZkClient.getTopicPartitionStatesRaw(partitions)) .thenReturn(Seq(GetDataResponse(Code.OK, null, Some(partition), TopicPartitionStateZNode.encode(leaderIsrAndControllerEpoch), stat, ResponseMetadata(0, 0)))) - val leaderAndIsrAfterElection = leaderAndIsr.newLeaderAndIsr(otherBrokerId, List(otherBrokerId)) - val updatedLeaderAndIsr = leaderAndIsrAfterElection.withPartitionEpoch(2) - when(mockZkClient.updateLeaderAndIsr(Map(partition -> leaderAndIsrAfterElection), controllerEpoch, controllerContext.epochZkVersion)) + + partitionStateMachine.handleStateChanges(partitions, OnlinePartition, Option(PreferredLeaderElectionStrategy)) + verify(mockControllerBrokerRequestBatch).newBatch() + verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) + verify(mockZkClient).getTopicPartitionStatesRaw(any()) + assertEquals(OnlinePartition, partitionState(partition)) + } + + @Test + def testOnlinePartitionToOnlineTransitionWithControlledShutdownBrokerInIsr(): Unit = { + val otherBrokerId = brokerId + 1 + val assignment = ReplicaAssignment(Seq(brokerId, otherBrokerId)) + val initialLeaderAndIsr = LeaderAndIsr(otherBrokerId, List(brokerId, otherBrokerId)) + val expectedLeaderAndIsr = initialLeaderAndIsr.newLeaderAndIsr(otherBrokerId, List(otherBrokerId)) + testOnlineToOnlineTransitionControlledShutdown(assignment, initialLeaderAndIsr, expectedLeaderAndIsr) + } + + @Test + def testOnlinePartitionToOnlineTransitionWithControlledShutdownBrokerNotInIsr(): Unit = { + val otherBrokerId = brokerId + 1 + val assignment = ReplicaAssignment(Seq(brokerId, otherBrokerId)) + val initialLeaderAndIsr = LeaderAndIsr(otherBrokerId, List(otherBrokerId)) + val expectedLeaderAndIsr = initialLeaderAndIsr.newEpoch + testOnlineToOnlineTransitionControlledShutdown(assignment, initialLeaderAndIsr, expectedLeaderAndIsr) + } + + @Test + def testOnlinePartitionToOnlineTransitionWithControlledShutdownBrokerLeaderWithEligibleFailoverLeader(): Unit = { + val otherBrokerId = brokerId + 1 + val assignment = ReplicaAssignment(Seq(brokerId, otherBrokerId)) + val initialLeaderAndIsr = LeaderAndIsr(brokerId, List(brokerId, otherBrokerId)) + val expectedLeaderAndIsr = initialLeaderAndIsr.newLeaderAndIsr(otherBrokerId, List(otherBrokerId)) + testOnlineToOnlineTransitionControlledShutdown(assignment, initialLeaderAndIsr, expectedLeaderAndIsr) + } + + @Test + def testOnlinePartitionToOnlineTransitionWithControlledShutdownBrokerLeaderWithNoEligibleFailoverLeader(): Unit = { + val otherBrokerId = brokerId + 1 + val assignment = ReplicaAssignment(Seq(brokerId, otherBrokerId)) + val initialLeaderAndIsr = LeaderAndIsr(brokerId, List(brokerId)) + assertThrows(classOf[StateChangeFailedException], () => { + testOnlineToOnlineTransitionControlledShutdown(assignment, initialLeaderAndIsr, initialLeaderAndIsr) + }) + } + + private def testOnlineToOnlineTransitionControlledShutdown( + assignment: ReplicaAssignment, + initialLeaderAndIsr: LeaderAndIsr, + expectedLeaderAndIsr: LeaderAndIsr + ): Unit = { + val otherBrokerId = brokerId + 1 + controllerContext.setLiveBrokers(assignment.replicas.map { replicaId => + TestUtils.createBrokerAndEpoch(replicaId, "host", 9092 + replicaId) + }.toMap) + controllerContext.shuttingDownBrokerIds.add(brokerId) + controllerContext.updatePartitionFullReplicaAssignment(partition, assignment) + controllerContext.putPartitionState(partition, OnlinePartition) + val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(initialLeaderAndIsr, controllerEpoch) + controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch) + + val stat = new Stat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) + when(mockZkClient.getTopicPartitionStatesRaw(partitions)) + .thenReturn(Seq(GetDataResponse(Code.OK, null, Some(partition), + TopicPartitionStateZNode.encode(leaderIsrAndControllerEpoch), stat, ResponseMetadata(0, 0)))) + val updatedLeaderAndIsr = expectedLeaderAndIsr.withPartitionEpoch(initialLeaderAndIsr.partitionEpoch + 1) + when(mockZkClient.updateLeaderAndIsr(Map(partition -> expectedLeaderAndIsr), controllerEpoch, controllerContext.epochZkVersion)) .thenReturn(UpdateLeaderAndIsrResult(Map(partition -> Right(updatedLeaderAndIsr)), Seq.empty)) - partitionStateMachine.handleStateChanges(partitions, OnlinePartition, Option(ControlledShutdownPartitionLeaderElectionStrategy)) + val results = partitionStateMachine.handleStateChanges(partitions, OnlinePartition, Option(ControlledShutdownStrategy)) + val result = results.get(partition).getOrElse(fail("Missing partition result in result set")) + result match { + case Left(exception) => throw exception + case Right(resultLeaderAndIsr) => assertEquals(updatedLeaderAndIsr, resultLeaderAndIsr) + } + verify(mockControllerBrokerRequestBatch).newBatch() - // The leaderAndIsr request should be sent to both brokers, including the shutting down one - verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(Seq(brokerId, otherBrokerId), - partition, LeaderIsrAndControllerEpoch(updatedLeaderAndIsr, controllerEpoch), replicaAssignment(Seq(brokerId, otherBrokerId)), - isNew = false) + // The LeaderAndIsr request should be sent only to the other broker + verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(Seq(otherBrokerId), + partition, LeaderIsrAndControllerEpoch(updatedLeaderAndIsr, controllerEpoch), assignment, isNew = false) + // The shutting down broker should see a StopReplica request + verify(mockControllerBrokerRequestBatch).addStopReplicaRequestForBrokers(Seq(brokerId), + partition, deletePartition = false) verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) verify(mockZkClient).getTopicPartitionStatesRaw(any()) verify(mockZkClient).updateLeaderAndIsr(any(), anyInt(), anyInt()) @@ -266,7 +338,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(false)) + Option(OfflineLeaderElectionStrategy(false)) ) verify(mockControllerBrokerRequestBatch).newBatch() verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(Seq(brokerId), @@ -343,7 +415,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(true)) + Option(OfflineLeaderElectionStrategy(true)) ) verify(mockControllerBrokerRequestBatch).newBatch() verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers( @@ -373,7 +445,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(false)) + Option(OfflineLeaderElectionStrategy(false)) ) verify(mockControllerBrokerRequestBatch).newBatch() verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) @@ -398,7 +470,7 @@ class PartitionStateMachineTest { partitionStateMachine.handleStateChanges( partitions, OnlinePartition, - Option(OfflinePartitionLeaderElectionStrategy(false)) + Option(OfflineLeaderElectionStrategy(false)) ) verify(mockControllerBrokerRequestBatch).newBatch() verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) @@ -469,7 +541,7 @@ class PartitionStateMachineTest { assertEquals(partitions.size, controllerContext.offlinePartitionCount, s"There should be ${partitions.size} offline partition(s)") - partitionStateMachine.handleStateChanges(partitions, OnlinePartition, Some(OfflinePartitionLeaderElectionStrategy(false))) + partitionStateMachine.handleStateChanges(partitions, OnlinePartition, Some(OfflineLeaderElectionStrategy(false))) assertEquals(0, controllerContext.offlinePartitionCount, s"There should be no offline partition(s)") } diff --git a/core/src/test/scala/unit/kafka/controller/ReplicaStateMachineTest.scala b/core/src/test/scala/unit/kafka/controller/ReplicaStateMachineTest.scala index 34187b138427f..b051a7bb031ed 100644 --- a/core/src/test/scala/unit/kafka/controller/ReplicaStateMachineTest.scala +++ b/core/src/test/scala/unit/kafka/controller/ReplicaStateMachineTest.scala @@ -30,9 +30,8 @@ import org.apache.zookeeper.KeeperException.Code import org.apache.zookeeper.data.Stat import org.junit.jupiter.api.Assertions._ import org.junit.jupiter.api.{BeforeEach, Test} -import org.mockito.ArgumentMatchers -import org.mockito.ArgumentMatchers.{any, anyInt} import org.mockito.Mockito.{mock, verify, when} +import org.mockito.ArgumentMatchers class ReplicaStateMachineTest { private var controllerContext: ControllerContext = null @@ -202,35 +201,78 @@ class ReplicaStateMachineTest { } @Test - def testOnlineReplicaToOfflineReplicaTransition(): Unit = { + def testOnlineReplicaToOfflineReplicaTransitionWithReplicaLeader(): Unit = { val otherBrokerId = brokerId + 1 val replicaIds = List(brokerId, otherBrokerId) + val initialLeaderAndIsr = LeaderAndIsr(brokerId, replicaIds) + val expectedLeaderAndIsr = initialLeaderAndIsr.newLeaderAndIsr(LeaderAndIsr.NoLeader, List(otherBrokerId)) + testOnlineReplicaToOfflineReplicaTransition(replicaIds, initialLeaderAndIsr, expectedLeaderAndIsr) + } + + @Test + def testOnlineReplicaToOfflineReplicaTransitionWithReplicaInIsr(): Unit = { + val otherBrokerId1 = brokerId + 1 + val otherBrokerId2 = brokerId + 2 + val replicaIds = List(brokerId, otherBrokerId1, otherBrokerId2) + val initialLeaderAndIsr = LeaderAndIsr(otherBrokerId1, replicaIds) + val expectedLeaderAndIsr = initialLeaderAndIsr.newLeaderAndIsr(otherBrokerId1, List(otherBrokerId1, otherBrokerId2)) + testOnlineReplicaToOfflineReplicaTransition(replicaIds, initialLeaderAndIsr, expectedLeaderAndIsr) + } + + @Test + def testOnlineReplicaToOfflineReplicaTransitionWhenReplicaOutOfIsr(): Unit = { + val otherBrokerId1 = brokerId + 1 + val otherBrokerId2 = brokerId + 2 + val replicaIds = List(brokerId, otherBrokerId1, otherBrokerId2) + val initialLeaderAndIsr = LeaderAndIsr(otherBrokerId1, List(otherBrokerId1)) + testOnlineReplicaToOfflineReplicaTransition(replicaIds, initialLeaderAndIsr, initialLeaderAndIsr) + } + + private def testOnlineReplicaToOfflineReplicaTransition( + replicaIds: Seq[Int], + initialLeaderAndIsr: LeaderAndIsr, + expectedLeaderAndIsr: LeaderAndIsr + ): Unit = { controllerContext.putReplicaState(replica, OnlineReplica) controllerContext.updatePartitionFullReplicaAssignment(partition, ReplicaAssignment(replicaIds)) - val leaderAndIsr = LeaderAndIsr(brokerId, replicaIds) - val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(leaderAndIsr, controllerEpoch) + val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(initialLeaderAndIsr, controllerEpoch) controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch) val stat = new Stat(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0) - val adjustedLeaderAndIsr = leaderAndIsr.newLeaderAndIsr(LeaderAndIsr.NoLeader, List(otherBrokerId)) - val updatedLeaderAndIsr = adjustedLeaderAndIsr.withPartitionEpoch(adjustedLeaderAndIsr.partitionEpoch + 1) + val updatedLeaderAndIsr = expectedLeaderAndIsr.withPartitionEpoch(expectedLeaderAndIsr.partitionEpoch + 1) val updatedLeaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(updatedLeaderAndIsr, controllerEpoch) + when(mockZkClient.getTopicPartitionStatesRaw(partitions)).thenReturn( Seq(GetDataResponse(Code.OK, null, Some(partition), TopicPartitionStateZNode.encode(leaderIsrAndControllerEpoch), stat, ResponseMetadata(0, 0)))) - when(mockZkClient.updateLeaderAndIsr(Map(partition -> adjustedLeaderAndIsr), controllerEpoch, controllerContext.epochZkVersion)) - .thenReturn(UpdateLeaderAndIsrResult(Map(partition -> Right(updatedLeaderAndIsr)), Seq.empty)) + + if (initialLeaderAndIsr != expectedLeaderAndIsr) { + when(mockZkClient.updateLeaderAndIsr(Map(partition -> expectedLeaderAndIsr), controllerEpoch, controllerContext.epochZkVersion)) + .thenReturn(UpdateLeaderAndIsrResult(Map(partition -> Right(updatedLeaderAndIsr)), Seq.empty)) + } else { + when(mockZkClient.updateLeaderAndIsr(Map.empty, controllerEpoch, controllerContext.epochZkVersion)) + .thenReturn(UpdateLeaderAndIsrResult(Map.empty, Seq.empty)) + } replicaStateMachine.handleStateChanges(replicas, OfflineReplica) - verify(mockControllerBrokerRequestBatch).newBatch() - verify(mockControllerBrokerRequestBatch).addStopReplicaRequestForBrokers(ArgumentMatchers.eq(Seq(brokerId)), ArgumentMatchers.eq(partition), ArgumentMatchers.eq(false)) - verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(Seq(otherBrokerId), - partition, updatedLeaderIsrAndControllerEpoch, replicaAssignment(replicaIds), isNew = false) - verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) - verify(mockZkClient).getTopicPartitionStatesRaw(any()) - verify(mockZkClient).updateLeaderAndIsr(any(), anyInt(), anyInt()) - assertEquals(updatedLeaderIsrAndControllerEpoch, controllerContext.partitionLeadershipInfo(partition).get) assertEquals(OfflineReplica, replicaState(replica)) + + val otherReplicaIds = replicaIds.filterNot(_ == brokerId) + verify(mockControllerBrokerRequestBatch).newBatch() + + if (initialLeaderAndIsr != expectedLeaderAndIsr) { + verify(mockControllerBrokerRequestBatch).addStopReplicaRequestForBrokers( + ArgumentMatchers.eq(Seq(brokerId)), ArgumentMatchers.eq(partition), ArgumentMatchers.eq(false)) + verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(otherReplicaIds, + partition, updatedLeaderIsrAndControllerEpoch, replicaAssignment(replicaIds), isNew = false) + verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) + assertEquals(updatedLeaderIsrAndControllerEpoch, controllerContext.partitionLeadershipInfo(partition).get) + } else { + verify(mockControllerBrokerRequestBatch).addLeaderAndIsrRequestForBrokers(otherReplicaIds, + partition, leaderIsrAndControllerEpoch, replicaAssignment(replicaIds), isNew = false) + verify(mockControllerBrokerRequestBatch).sendRequestsToBrokers(controllerEpoch) + assertEquals(leaderIsrAndControllerEpoch, controllerContext.partitionLeadershipInfo(partition).get) + } } @Test diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala index ad3c34f9605aa..a8bf448c43416 100755 --- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala +++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala @@ -1093,9 +1093,9 @@ object TestUtils extends Logging { brokers: Seq[B], timeout: Long = JTestUtils.DEFAULT_MAX_WAIT_MS): Unit = { val expectedBrokerIds = brokers.map(_.config.brokerId).toSet - waitUntilTrue(() => brokers.forall(server => + waitUntilTrue(() => brokers.forall { server => expectedBrokerIds == server.dataPlaneRequestProcessor.metadataCache.getAliveBrokers().map(_.id).toSet - ), "Timed out waiting for broker metadata to propagate to all servers", timeout) + }, "Timed out waiting for broker metadata to propagate to all servers", timeout) } /** @@ -1889,16 +1889,22 @@ object TestUtils extends Logging { ) } + def currentIsr(admin: Admin, partition: TopicPartition): Set[Int] = { + val description = admin.describeTopics(Set(partition.topic).asJava) + .allTopicNames + .get + .asScala + description + .values + .flatMap(_.partitions.asScala.flatMap(_.isr.asScala)) + .map(_.id) + .toSet + } + def waitForBrokersInIsr(client: Admin, partition: TopicPartition, brokerIds: Set[Int]): Unit = { waitUntilTrue( () => { - val description = client.describeTopics(Set(partition.topic).asJava).allTopicNames.get.asScala - val isr = description - .values - .flatMap(_.partitions.asScala.flatMap(_.isr.asScala)) - .map(_.id) - .toSet - + val isr = currentIsr(client, partition) brokerIds.subsetOf(isr) }, s"Expected brokers $brokerIds to be in the ISR for $partition"