From f73e1640f0a0f9fe1937544a2cfe5d457f85483c Mon Sep 17 00:00:00 2001 From: Ron Dagostino Date: Mon, 15 Mar 2021 17:35:21 -0400 Subject: [PATCH 1/3] KAFKA-12455: OffsetValidationTest.test_broker_rolling_bounce fail: Raft --- .../kafka/server/RaftReplicaManager.scala | 18 ++++++++---------- tests/kafkatest/tests/client/consumer_test.py | 16 +++++++++------- 2 files changed, 17 insertions(+), 17 deletions(-) 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..1984c605ed16a 100644 --- a/tests/kafkatest/tests/client/consumer_test.py +++ b/tests/kafkatest/tests/client/consumer_test.py @@ -26,12 +26,14 @@ class OffsetValidationTest(VerifiableConsumerTest): TOPIC = "test_topic" NUM_PARTITIONS = 1 + DEFAULT_NUM_BROKERS = 2 def __init__(self, test_context): - super(OffsetValidationTest, self).__init__(test_context, num_consumers=3, num_producers=1, - num_zk=1, num_brokers=2, topics={ - self.TOPIC : { 'partitions': self.NUM_PARTITIONS, 'replication-factor': 2 } - }) + super(OffsetValidationTest, self).__init__(test_context, num_consumers=3, num_producers=1, num_zk=1, + num_brokers= self.DEFAULT_NUM_BROKERS if not test_context.injected_args \ + else test_context.injected_args.get('num_brokers', self.DEFAULT_NUM_BROKERS), + topics={self.TOPIC : { 'partitions': self.NUM_PARTITIONS, 'replication-factor': 2 }}, + ) def rolling_bounce_consumers(self, consumer, keep_alive=0, num_bounces=5, clean_shutdown=True): for _ in range(num_bounces): @@ -74,9 +76,9 @@ def setup_consumer(self, topic, **kwargs): self.mark_for_collect(consumer, 'verifiable_consumer_stdout') return consumer - @cluster(num_nodes=7) - @matrix(metadata_quorum=quorum.all_non_upgrade) - def test_broker_rolling_bounce(self, metadata_quorum=quorum.zk): + @cluster(num_nodes=8) + @matrix(num_brokers=[3], metadata_quorum=quorum.all_non_upgrade) + def test_broker_rolling_bounce(self, num_brokers, metadata_quorum=quorum.zk): """ Verify correct consumer behavior when the brokers are consecutively restarted. From 1b9e25eab6ac54336901f691e48b4c79049679e7 Mon Sep 17 00:00:00 2001 From: Ron Dagostino Date: Mon, 15 Mar 2021 18:44:07 -0400 Subject: [PATCH 2/3] Fix via 30 second session timeout instead of 3 brokers --- tests/kafkatest/tests/client/consumer_test.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/tests/kafkatest/tests/client/consumer_test.py b/tests/kafkatest/tests/client/consumer_test.py index 1984c605ed16a..1d7d76943baae 100644 --- a/tests/kafkatest/tests/client/consumer_test.py +++ b/tests/kafkatest/tests/client/consumer_test.py @@ -26,14 +26,12 @@ class OffsetValidationTest(VerifiableConsumerTest): TOPIC = "test_topic" NUM_PARTITIONS = 1 - DEFAULT_NUM_BROKERS = 2 def __init__(self, test_context): - super(OffsetValidationTest, self).__init__(test_context, num_consumers=3, num_producers=1, num_zk=1, - num_brokers= self.DEFAULT_NUM_BROKERS if not test_context.injected_args \ - else test_context.injected_args.get('num_brokers', self.DEFAULT_NUM_BROKERS), - topics={self.TOPIC : { 'partitions': self.NUM_PARTITIONS, 'replication-factor': 2 }}, - ) + super(OffsetValidationTest, self).__init__(test_context, num_consumers=3, num_producers=1, + num_zk=1, num_brokers=2, topics={ + self.TOPIC : { 'partitions': self.NUM_PARTITIONS, 'replication-factor': 2 } + }) def rolling_bounce_consumers(self, consumer, keep_alive=0, num_bounces=5, clean_shutdown=True): for _ in range(num_bounces): @@ -76,9 +74,9 @@ def setup_consumer(self, topic, **kwargs): self.mark_for_collect(consumer, 'verifiable_consumer_stdout') return consumer - @cluster(num_nodes=8) - @matrix(num_brokers=[3], metadata_quorum=quorum.all_non_upgrade) - def test_broker_rolling_bounce(self, num_brokers, metadata_quorum=quorum.zk): + @cluster(num_nodes=7) + @matrix(metadata_quorum=quorum.all_non_upgrade) + def test_broker_rolling_bounce(self, metadata_quorum=quorum.zk): """ Verify correct consumer behavior when the brokers are consecutively restarted. @@ -95,6 +93,7 @@ def test_broker_rolling_bounce(self, num_brokers, metadata_quorum=quorum.zk): partition = TopicPartition(self.TOPIC, 0) producer = self.setup_producer(self.TOPIC) + self.session_timeout_sec = 30 consumer = self.setup_consumer(self.TOPIC) producer.start() From 4bcf768dc4a2089a1943193b044f016d73ae590f Mon Sep 17 00:00:00 2001 From: Ron Dagostino Date: Tue, 16 Mar 2021 10:46:33 -0400 Subject: [PATCH 3/3] Add comment --- tests/kafkatest/tests/client/consumer_test.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/kafkatest/tests/client/consumer_test.py b/tests/kafkatest/tests/client/consumer_test.py index 1d7d76943baae..49e9331f613f9 100644 --- a/tests/kafkatest/tests/client/consumer_test.py +++ b/tests/kafkatest/tests/client/consumer_test.py @@ -93,6 +93,13 @@ 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)