[LI-HOTFIX] Avoid assigning replicas to the preferred controllers or maintenance brokers

[gitlw]

If we expand partitions using the AdminZkClient, currently there is no logic to avoid assigning replicas to preferred controllers.
This PR fixes the issue in the following two places:

1. It avoids assignment of replicas to preferred controllers and maintenance brokers within the AdminZkClient
2. In case someone is using an non-patched client and still assigns replicas to preferred controllers, the KafkaController will call rearrangePartitionReplicaAssignmentForNewPartitions to remove the preferred controllers from the assignment.

Testing Stretegy:
An unit test is added to ensure the problem doesn't happen again.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[xiowu0]

could we add some safe check here to ensure the partition state znode doesn't exist for these new partitions. Although I don't see an issue in the current implementation, if this function is used incorrectly, it would be very dangerous. For example, if we use "rearrangePartitionReplicaAssignmentForNewPartitions(topics, false)" when initializing controller context, this may cause all topics to get reassigned since controllerContext.partitionAssignments is an empty set (this will also result in orphan partitions).
Again, I don't see the issue in this implementation, but I think it is worth checking.

In addition, this doesn't address the case when there is controller move right after partition expansion (before the rearrange actually happened). I think it is ok since we are not trying to handle 100% no replicas in the preferred controller. Maybe we can add some comments for this unhandled situations.

[gitlw]

Thanks for the comments. It's true that they are all newPartitions given controllerContext.partitionAssignments is an empty set, but they shouldn't have a replica on the noNewPartitionBrokers, thus they shouldn't be rearranged.

It's a good point that the current implementation cannot handle controller switches. Given it's safe to scan all topics' partitions on controller switch, I think it should be done during a controller switch. Thoughts?

[xiowu0]

manual assignment for existing topics can still assign partitions to noNewPartitionBrokers. I think we cannot guarantee that noNewPartitionBrokers won't get replicas. In addition, there are some small-time window that new replicas can still get assigned to preferred controllers due to the fact that preferred controller znode is emphermal znode (see the design doc for more detail).

I think a safer way is to rely on the existence of partition state znode when performing rearrangement.

scan all topics' partitions on controller switch => If we cannot guarantee 100% no replica in the preferred controllers, I think it is ok to not performing special handing during controller switch given the overhead/additional hacky code needed (up to you to make a decision).

[gitlw]

Upon closer look, I find that there is already logic to handle the controller switch over by calling the rearrangePartitionReplicaAssignmentForNewPartitions method inside initializeControllerContext.

Regarding the preferred controller znodes being ephemeral, it's kinda an orthogonal design issue that we could address independently.

[xiowu0]

does it conflict with partition reassignment?
say if a partition gets reassigned to replica ( 1, 2, 5, 6 ) ==> if one of this replica is maintenance brokers, would this reassignment complete if we automatically changing zk node to disallow placing replicas on maintenance brokers.

[gitlw]

That's a good point. I've updated the PR to avoid assignment replicas to the undesirable hosts during partition reassignment. Please take another look. Thanks!

[xiowu0]

still have some race condition here, because a broker can be marked as maintenance brokers after partition reassignment request is received before the reassignment request is completed

[efeg]

Quick Update on Review: Had an offline chat with @gitlw regarding whether we may want to address the second point (i.e. 2. In case someone is using an non-patched client and still assigns replicas to preferred controllers, the KafkaController will call rearrangePartitionReplicaAssignmentForNewPartitions to remove the preferred controllers from the assignment.) in another way. We will chat more about the pros and cons (cc @xiowu0).