diff --git a/core/src/main/scala/kafka/server/RaftReplicaManager.scala b/core/src/main/scala/kafka/server/RaftReplicaManager.scala index 143709deefe86..37dc3e1a23ec3 100644 --- a/core/src/main/scala/kafka/server/RaftReplicaManager.scala +++ b/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. diff --git a/tests/kafkatest/tests/client/consumer_test.py b/tests/kafkatest/tests/client/consumer_test.py index f41748078ca72..49e9331f613f9 100644 --- a/tests/kafkatest/tests/client/consumer_test.py +++ b/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()