Removing entries from the shuttingdownBrokerIds upon broker startup

[gitlw]

Consider a cluster having two brokers (id=0 and id=1) configured with

<property name="kafka.controlledShutdownEnable" value="true">
<property name="kafka.controlledShutdownMaxRetries" value="2147483640">
<property name="kafka.controlledShutdownSafetyCheckEnable" value="true">
<property name="kafka.controlledShutdownSafetyCheckRedundancyFactor" value="0">

and a partition having two replicas on these two brokers.

If broker1 goes through the controlled shutdown process and starts up, the current implementation uses the following logic to update the shuttingdownBrokerIds:

shuttingDownBrokerIds = shuttingDownBrokerIds.filter(b =>
      liveBrokerEpochs.contains(b._1) && b._2 < liveBrokerEpochs(b._1))

which means broker1's previous shutting epoch is still retained in the shuttingDownBrokerIds after it restarts.

Later on, if broker0 needs to restart, the controller uses the leaderForControlledShutdown method to elect a new leader for partitions led by broker0:

def leaderForControlledShutdown(controllerContext: ControllerContext,
                                  leaderAndIsrs: Seq[(TopicPartition, LeaderAndIsr)]): Seq[ElectionResult] = {
    val shuttingDownBrokerIdSet = controllerContext.shuttingDownBrokerIds.keySet.toSet
    leaderAndIsrs.map { case (partition, leaderAndIsr) =>
      leaderForControlledShutdown(partition, leaderAndIsr, shuttingDownBrokerIdSet, controllerContext)
    }
  }

The leaderForControlledShutdown would still treat broker1 as a shutting down broker, and therefore won't be able to transfer the leadership to broker1, despite its readiness to serve as the leader.

This PR resolves the problem by changing the update logic of shuttingDownBrokerIds such that the entry of a restarted broker would be removed from it.

Committer Checklist (excluded from commit message)

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

[gitlw]

The failed tests are being resolved at #188

[ambroff]

+1. Thanks for fixing this. That's a good catch, I just had the logic wrong in that filter predicate.

scala> val liveBrokerEpochs = Map(0 -> 5, 1 -> 10)
val liveBrokerEpochs: scala.collection.immutable.Map[Int,Int] = Map(0 -> 5, 1 -> 10)

scala> liveBrokerEpochs(0)
val res0: Int = 5

scala> val shuttingDownBrokerIds = Map(1 -> 9)
val shuttingDownBrokerIds: scala.collection.immutable.Map[Int,Int] = Map(1 -> 9)

scala> shuttingDownBrokerIds.filter(b => liveBrokerEpochs.contains(b._1) && b._2 < liveBrokerEpochs(b._1))
val res1: scala.collection.immutable.Map[Int,Int] = Map(1 -> 9)

scala> shuttingDownBrokerIds.filter(b => liveBrokerEpochs.contains(b._1) && b._2 > liveBrokerEpochs(b._1))
val res2: scala.collection.immutable.Map[Int,Int] = Map()

scala> shuttingDownBrokerIds.filter(b => liveBrokerEpochs.contains(b._1) && b._2 > liveBrokerEpochs(b._1))
val res3: scala.collection.immutable.Map[Int,Int] = Map()

scala> shuttingDownBrokerIds.filter(b => liveBrokerEpochs.contains(b._1) && b._2 > liveBrokerEpochs(b._1))
val res4: scala.collection.immutable.Map[Int,Int] = Map()