soarez commented on code in PR #15557:
URL: https://github.com/apache/kafka/pull/15557#discussion_r1546234986
##########
core/src/main/scala/kafka/server/ReplicaAlterLogDirsThread.scala:
##########
@@ -96,57 +97,55 @@ class ReplicaAlterLogDirsThread(name: String,
}
override def removePartitions(topicPartitions: Set[TopicPartition]):
Map[TopicPartition, PartitionFetchState] = {
- // Schedule assignment request to revert any queued request before
cancelling
- for {
- topicPartition <- topicPartitions
- partitionState <- partitionAssignmentRequestState(topicPartition)
- if partitionState == QUEUED
- partition = replicaMgr.getPartitionOrException(topicPartition)
- topicId <- partition.topicId
- directoryId <- partition.logDirectoryId()
- topicIdPartition = new TopicIdPartition(topicId,
topicPartition.partition())
- } directoryEventHandler.handleAssignment(topicIdPartition, directoryId, ()
=> ())
+ for (topicPartition <- topicPartitions) {
+ // Revert any reassignments for partitions that did not complete the
future replica promotion
+ val PromotionState(reassignmentState, topicId, originalDir) =
this.promotionState(topicPartition)
+ if (reassignmentState.inconsistentMetadata) {
+ directoryEventHandler.handleAssignment(new TopicIdPartition(topicId,
topicPartition.partition()), originalDir, () => ())
Review Comment:
Good questions.
1. Yes. See `ReplicationController.handleAssignReplicasToDirs`. If a replica
is assigned to a directory that is not an online directory in the broker,
leadership and ISR is updated.
2. Yes, I'll update the KIP. At the time, I thought it made more sense to
use `UUID.LOST_DIR`, but in practice it's a more general solution to just
revert the assignment.
3. I've started but haven't yet been able to come up with an integration
test for this. It requires blocking the process of replica fetching and
assginment processing and injecting failures at very specific times. Let me
know if you think it's critical to have an integration test for this, and I'll
give it another go.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
