KAFKA-8171: Set callback to null in addStopReplicaRequestForBrokers when replica …

[kehuum]

Problem:

In ControllerChannelManager.sendRequestsToBrokers, for STOP_REPLICA requests, it will try to group the requests based on deletePartition flag and callback:

val (replicasToGroup, replicasToNotGroup) = replicaInfoList.partition(r => !r.deletePartition && r.callback == null)

When both conditions meet, controller is expected to only send one request to destionation broker. However, when adding the requests in ReplicaStateMachine, it's putting in non-null callback (,)=>(). Therefore, replicasToGroup is always empty and controller will always first sends an empty request followed by #partitions requests.

Fix: set the callback to null in addStopReplicaRequestForBrokers when replica state transits to offline.

Committer Checklist (excluded from commit message)

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

[hzxa21]

@kehuum Thanks for the patch. Can you also add a unit test for this to make sure we are actually actually doing the batching?

[ijuma]

Nice catch. Relying on a null callback is too brittle, however. One option is to make the callback parameter an Option. Even better would be to make it impossible to make this kind of mistake. I haven't looked at the code in detail to see if that's possible though.

[hachikuji]

We ended up fixing this in #6588 as part of removing the callback mechanism entirely. I will go ahed and close this. Please reopen if I missed something.