KAFKA-5107: remove preferred replica election state from ControllerContext #2927
Conversation
…ntext KAFKA-5028 moves the controller to a single-threaded model, so we would no longer have work interleaved between preferred replica leader election, meaning we don't need to keep its state. This patch additionally addresses a bug from KAFKA-5028 where it made onPreferredReplicaElection keep the line calling topicDeletionManager.markTopicIneligibleForDeletion but removes the line calling topicDeletionManager.resumeDeletionForTopics
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
There was a problem hiding this comment.
@onurkaraman : Thanks for the patch. Left a couple of comments.
| val topicsWithReplicasOnDeadBrokers = controllerContext.partitionReplicaAssignment.filter { case (_, replicas) => | ||
| replicas.exists(r => !controllerContext.liveBrokerIds.contains(r)) }.keySet.map(_.topic) | ||
| val topicsForWhichPreferredReplicaElectionIsInProgress = controllerContext.partitionsUndergoingPreferredReplicaElection.map(_.topic) | ||
| val topicsForWhichPreferredReplicaElectionIsInProgress = pendingPreferredReplicaElections.map(_.topic) |
There was a problem hiding this comment.
A couple of comments. (1) I think topic deletion should have higher priority than preferred replication election. So, if a topic is marked for deletion, we should stop all pending preferred leader election, instead of letting that block the topic deletion. (2) It seems that we don't need to include topicsWithReplicasOnDeadBrokers in topicsIneligibleForDeletion. The same logic is already done in TopicDeletionManager.startReplicaDeletion() when the topic deletion is resumed.
There was a problem hiding this comment.
fixed (1) but I'm not sure about the impact of (2) so I left it as is for now.
| @@ -248,6 +247,8 @@ class KafkaController(val config: KafkaConfig, zkUtils: ZkUtils, val brokerState | |||
|
|
|||
| initializeControllerContext() | |||
|
|
|||
| val pendingPreferredReplicaElections = fetchPendingPreferredReplicaElections() | |||
| val (topicsToBeDeleted, topicsIneligibleForDeletion) = fetchTopicDeletionsInProgress(pendingPreferredReplicaElections) | |||
There was a problem hiding this comment.
I think this probably needs to be computed after maybeTriggerPartitionReassignment(), which may complete some of the partition reassignment.
There was a problem hiding this comment.
This PR's refactoring of TopicDeletionManager's construction may have complicated things a bit.
Positioning these lines is a bit tricky. We want to load up topic deletion state into TopicDeletionManager prior to calling PartitionStateMachine.startup, as triggerOnlinePartitionStateChange checks if !controller.topicDeletionManager.isTopicQueuedUpForDeletion(topicAndPartition.topic).
There was a problem hiding this comment.
Good point. One way to do that is to de-couple the initialization of the state in TopicDeletionManager from the handling of topic deletion. So, the sequence will be
(1) TopicDeletionManager.init(topicsToBeDeleted, topicsIneligibleForDeletion) // don't call resumeDeletions.
(2) replicaStateMachine.startup()
(3) partitionStateMachine.startup()
(4) maybeTriggerPartitionReassignment()
(5) topicDeletionManager.resumeDeletions().
There was a problem hiding this comment.
This is exactly the solution I had been working on. Glad to see we're on the same page!
… and prioritize deletions over preferred replica leader elections
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
There was a problem hiding this comment.
@onurkaraman : Thanks for the update. Just one minor comment.
| @@ -84,6 +83,12 @@ class TopicDeletionManager(controller: KafkaController) extends Logging { | |||
| } | |||
| } | |||
|
|
|||
| def start(): Unit = { | |||
There was a problem hiding this comment.
Having init(), start(), shutdown() together can be a bit confusing. How about rename them to init(), tryTopicDeletion() and reset()?
There was a problem hiding this comment.
completely agree. updated the pr.
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
retest this please |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
@onurkaraman : Thanks for the patch. LGTM |
KAFKA-5028 moves the controller to a single-threaded model, so we would no longer have work interleaved between preferred replica leader election, meaning we don't need to keep its state.
This patch additionally addresses a bug from KAFKA-5028 where it made onPreferredReplicaElection keep the line calling topicDeletionManager.markTopicIneligibleForDeletion but removes the line calling topicDeletionManager.resumeDeletionForTopics