KAFKA-12455: OffsetValidationTest.test_broker_rolling_bounce fail: Raft

core/src/main/scala/kafka/server/RaftReplicaManager.scala

@@ -132,8 +132,8 @@ class RaftReplicaManager(config: KafkaConfig,
 
   def endMetadataChangeDeferral(onLeadershipChange: (Iterable[Partition], Iterable[Partition]) => Unit): Unit = {
     val startMs = time.milliseconds()
-    val partitionsMadeFollower = mutable.Set[Partition]()
-    val partitionsMadeLeader = mutable.Set[Partition]()
+    var partitionsMadeFollower = Set.empty[Partition]
+    var partitionsMadeLeader = Set.empty[Partition]
     replicaStateChangeLock synchronized {
       stateChangeLogger.info(s"Applying deferred metadata changes")
       val highWatermarkCheckpoints = new LazyOffsetCheckpoints(this.highWatermarkCheckpoints)
@@ -156,14 +156,10 @@ class RaftReplicaManager(config: KafkaConfig,
           }
         }
 
-        val partitionsMadeLeader = if (leaderPartitionStates.nonEmpty)
-          delegate.makeLeaders(partitionsAlreadyExisting, leaderPartitionStates, highWatermarkCheckpoints, None)
-        else
-          Set.empty[Partition]
-        val partitionsMadeFollower = if (followerPartitionStates.nonEmpty)
-          delegate.makeFollowers(partitionsAlreadyExisting, brokers, followerPartitionStates, highWatermarkCheckpoints, None)
-        else
-          Set.empty[Partition]
+        if (leaderPartitionStates.nonEmpty)
+          partitionsMadeLeader = delegate.makeLeaders(partitionsAlreadyExisting, leaderPartitionStates, highWatermarkCheckpoints, None)
+        if (followerPartitionStates.nonEmpty)
+          partitionsMadeFollower = delegate.makeFollowers(partitionsAlreadyExisting, brokers, followerPartitionStates, highWatermarkCheckpoints, None)
 
         // We need to transition anything that hasn't transitioned from Deferred to Offline to the Online state.
         deferredPartitionsIterator.foreach { deferredPartition =>
@@ -331,6 +327,8 @@ class RaftReplicaManager(config: KafkaConfig,
         replicaFetcherManager.shutdownIdleFetcherThreads()
         replicaAlterLogDirsManager.shutdownIdleFetcherThreads()
         onLeadershipChange(partitionsBecomeLeader, partitionsBecomeFollower)
+        stateChangeLogger.info(s"Metadata batch $metadataOffset: applied ${partitionsBecomeLeader.size + partitionsBecomeFollower.size} partitions: " +
+          s"${partitionsBecomeLeader.size} leader(s) and ${partitionsBecomeFollower.size} follower(s)")
       }
       // TODO: we should move aside log directories which have been deleted rather than
       // purging them from the disk immediately.

tests/kafkatest/tests/client/consumer_test.py

@@ -93,6 +93,14 @@ def test_broker_rolling_bounce(self, metadata_quorum=quorum.zk):
         partition = TopicPartition(self.TOPIC, 0)
 
         producer = self.setup_producer(self.TOPIC)
+        # The consumers' session timeouts must exceed the time it takes for a broker to roll.  Consumers are likely
+        # to see cluster metadata consisting of just a single alive broker in the case where the cluster has just 2
+        # brokers and the cluster is rolling (which is what is happening here).  When the consumer sees a single alive
+        # broker, and then that broker rolls, the consumer will be unable to connect to the cluster until that broker
+        # completes its roll.  In the meantime, the consumer group will move to the group coordinator on the other
+        # broker, and that coordinator will fail the consumer and trigger a group rebalance if its session times out.
+        # This test is asserting that no rebalances occur, so we increase the session timeout for this to be the case.
+        self.session_timeout_sec = 30
         consumer = self.setup_consumer(self.TOPIC)
 
         producer.start()