Skip to content

KAFKA-13972; Ensure that replicas are stopped after cancelled reassignment #12271

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
hachikuji wants to merge 8 commits into from
Closed

KAFKA-13972; Ensure that replicas are stopped after cancelled reassignment #12271

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
04fde73
KAFKA-13972; Ensure that replicas are stopped after cancelled reassig…
hachikuji Jun 8, 2022
34688e0
Stop replicas during partition state transitions
hachikuji Jun 13, 2022
1dfa9a6
Fix test failures and build issues
hachikuji Jun 13, 2022
ad06ee3
Add missing header to ElectionTest
hachikuji Jun 13, 2022
41366ef
Fixing/improving test cases
hachikuji Jun 14, 2022
cff261d
more cleanups and test fixes
hachikuji Jun 14, 2022
1f92c67
Remove unused reassign leader logic
hachikuji Jun 14, 2022
266baae
Add some documentation for reassignment cancellation/completion
hachikuji Jun 14, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/controller/KafkaController.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2653,7 +2653,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]"
}
}

Expand Down
100 changes: 61 additions & 39 deletions core/src/main/scala/kafka/controller/ReplicaStateMachine.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -218,22 +218,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))
val updatedLeaderIsrAndControllerEpochs = fenceReplicas(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)
Expand All @@ -245,7 +251,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 =>
Expand Down Expand Up @@ -286,27 +300,27 @@ class ZkReplicaStateMachine(config: KafkaConfig,
}

/**
* Repeatedly attempt to remove a replica from the isr of multiple partitions until there are no more remaining partitions
* Repeatedly attempt to fence a replica of multiple partitions until there are no more remaining partitions
* to retry.
* @param replicaId The replica being removed from isr of multiple partitions
* @param partitions The partitions from which we're trying to remove the replica from isr
* @return The updated LeaderIsrAndControllerEpochs of all partitions for which we successfully removed the replica from isr.
* @param replicaId The replica being fenced from multiple partitions
* @param partitions The partitions from which we're trying to fence the replica from
* @return The updated LeaderIsrAndControllerEpochs of all partitions for which we successfully fenced
*/
private def removeReplicasFromIsr(
private def fenceReplicas(
replicaId: Int,
partitions: Seq[TopicPartition]
): Map[TopicPartition, LeaderIsrAndControllerEpoch] = {
var results = Map.empty[TopicPartition, LeaderIsrAndControllerEpoch]
var remaining = partitions
while (remaining.nonEmpty) {
val (finishedRemoval, removalsToRetry) = doRemoveReplicasFromIsr(replicaId, remaining)
remaining = removalsToRetry
var remainingPartitions = partitions
while (remainingPartitions.nonEmpty) {
val (finishedPartitions, partitionsToRetry) = tryFenceReplicas(replicaId, remainingPartitions)
remainingPartitions = partitionsToRetry

finishedRemoval.foreach {
finishedPartitions.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
}
Expand All @@ -315,35 +329,43 @@ class ZkReplicaStateMachine(config: KafkaConfig,
}

/**
* Try to remove a replica from the isr of multiple partitions.
* Removing a replica from isr updates partition state in zookeeper.
* Fence an existing replica by bumping the leader epoch. The bumped leader epoch
* ensures 1) that the follower can no longer fetch with the old epoch, and 2)
* that it will accept a `StopReplica` request with the bumped epoch.
*
* @param replicaId The replica being removed from isr of multiple partitions
* @param partitions The partitions from which we're trying to remove the replica from isr
* - If the replica is the current leader, then the leader will be changed to
* [[LeaderAndIsr.NoLeader]], and it will remain in the ISR.
* - If the replica is not the current leader and it is in the ISR, then it
* will be removed from the ISR.
* - Otherwise, the epoch will be bumped and the leader and ISR will be unchanged.
*
* Fencing a replica updates partition state in zookeeper.
*
* @param replicaId The replica being fenced from multiple partitions
* @param partitions The partitions from which we're trying to fence the replica from
* @return A tuple of two elements:
* 1. The updated Right[LeaderIsrAndControllerEpochs] of all partitions for which we successfully
* removed the replica from isr. Or Left[Exception] corresponding to failed removals that should
* not be retried
* fenced the replica. Or Left[Exception] for failures which need to be retries
* 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 tryFenceReplicas(
replicaId: Int,
partitions: Seq[TopicPartition]
): (Map[TopicPartition, Either[Exception, LeaderIsrAndControllerEpoch]], Seq[TopicPartition]) = {
val (leaderAndIsrs, partitionsWithNoLeaderAndIsrInZk) = getTopicPartitionStatesFromZk(partitions)
val (leaderAndIsrsWithReplica, leaderAndIsrsWithoutReplica) = leaderAndIsrs.partition { case (_, result) =>
result.map { leaderAndIsr =>
leaderAndIsr.isr.contains(replicaId)
}.getOrElse(false)
}

val adjustedLeaderAndIsrs: Map[TopicPartition, LeaderAndIsr] = leaderAndIsrsWithReplica.flatMap {
val adjustedLeaderAndIsrs: Map[TopicPartition, LeaderAndIsr] = leaderAndIsrs.flatMap {
case (partition, result) =>
result.toOption.map { leaderAndIsr =>
val newLeader = if (replicaId == leaderAndIsr.leader) LeaderAndIsr.NoLeader else leaderAndIsr.leader
val adjustedIsr = if (leaderAndIsr.isr.size == 1) leaderAndIsr.isr else leaderAndIsr.isr.filter(_ != replicaId)
partition -> leaderAndIsr.newLeaderAndIsr(newLeader, adjustedIsr)
if (leaderAndIsr.isr.contains(replicaId)) {
val newLeader = if (replicaId == leaderAndIsr.leader) LeaderAndIsr.NoLeader else leaderAndIsr.leader
val adjustedIsr = if (leaderAndIsr.isr.size == 1) leaderAndIsr.isr else leaderAndIsr.isr.filter(_ != replicaId)
partition -> leaderAndIsr.newLeaderAndIsr(newLeader, adjustedIsr)
} else {
// Even if the replica is not in the ISR. We must bump the epoch to ensure the replica
// is fenced from replication and the `StopReplica` can be sent with a bumped epoch.
partition -> leaderAndIsr.newEpoch

@junrao junrao Jun 9, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A few comments on this.
(1) This seems to be a too low level to do this. When we cancel a reassignment, we may remove multiple replicas from a partition. Instead of bumping up the leader epoch for every replica, it's probably better to bump up the leader epoch once. Perhaps it's better to do this at the high level in KafkaController.updateCurrentReassignment().
(2) Controlled shutdown also goes through Offline transition. It's possible that the shutting down broker is out of isr. In that case, do we want to bump up the leader epoch?
(3) The current controlled shutdown logic also seems to have the same issue that its StopReplicaRequest will be ignored. When doing a controlled shutdown for a follower, it seems that we don't bump up the leader epoch. So, the StopReplicaRequest will be ignored by the follower?
(4) Should we do the same fix in KRaft?

@hachikuji hachikuji Jun 9, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@junrao Thanks for the comments. I think you are right about this issue affecting ControlledShutdown as well. I have been trying to understand that path a little better since it looks like it is not doing quite what we (or at least I) expect today. Also agree about the epoch bump being too low-level. We might need the fencing of replicas to happen at the partition level instead of the replica level. Let me take a look at this.

Interestingly, I don't think the issue affects KRaft. We bump the leader epoch in KRaft any time a replica is removed from the ISR or the replica set.

}
}
}

Expand All @@ -362,13 +384,13 @@ class ZkReplicaStateMachine(config: KafkaConfig,
}.toMap

val leaderIsrAndControllerEpochs: Map[TopicPartition, Either[Exception, LeaderIsrAndControllerEpoch]] =
(leaderAndIsrsWithoutReplica ++ finishedPartitions).map { case (partition, result) =>
finishedPartitions.map { case (partition, result) =>
(partition, result.map { leaderAndIsr =>
val leaderIsrAndControllerEpoch = LeaderIsrAndControllerEpoch(leaderAndIsr, controllerContext.epoch)
controllerContext.putPartitionLeadershipInfo(partition, leaderIsrAndControllerEpoch)
leaderIsrAndControllerEpoch
})
}
}.toMap

if (isDebugEnabled) {
updatesToRetry.foreach { partition =>
Expand Down
95 changes: 77 additions & 18 deletions core/src/test/scala/integration/kafka/admin/ReassignPartitionsIntegrationTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,12 +108,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))
Expand All @@ -122,10 +123,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)
Expand All @@ -137,6 +136,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)
Expand Down Expand Up @@ -296,10 +299,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"]}""" +
Expand All @@ -314,14 +320,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.
Expand All @@ -330,6 +333,62 @@ 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)
}

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 = {
Expand Down Expand Up @@ -541,8 +600,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}, " +
Expand Down
Loading