Add wait to prevent race condition in testAutoKafkaRebalanceScaleUpSc…

…aleDown Signed-off-by: see-quick <[email protected]>
strimzi · Nov 14, 2024 · 32b3364 · 32b3364
1 parent 3ecb5b9
commit 32b3364
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 4 deletions.
diff --git a/systemtest/src/main/java/io/strimzi/systemtest/utils/kafkaUtils/KafkaTopicUtils.java b/systemtest/src/main/java/io/strimzi/systemtest/utils/kafkaUtils/KafkaTopicUtils.java
@@ -248,6 +248,33 @@ public static List<String> getKafkaTopicReplicasForEachPartition(String namespac
             .collect(Collectors.toList());
     }
 
+    /**
+     * Waits for the topic replicas to be moved to the specified brokers.
+     *
+     * @param namespaceName   The Kubernetes namespace in which the KafkaTopic resides.
+     * @param topicName       The name of the KafkaTopic.
+     * @param scraperPodName  The name of the pod used to scrape Kafka metrics.
+     * @param bootstrapServer The bootstrap server address.
+     * @param brokerIds       A list of broker IDs where replicas should be moved.
+     */
+    public static void waitForTopicReplicasOnBrokers(String namespaceName, String topicName, String scraperPodName, String bootstrapServer, List<String> brokerIds) {
+        LOGGER.info("Waiting for topic replicas of {} to be moved to brokers {}", topicName, brokerIds);
+        TestUtils.waitFor(
+            "Wait for topic replicas to be on the specified brokers",
+            TestConstants.GLOBAL_POLL_INTERVAL,
+            TestConstants.GLOBAL_TIMEOUT,
+            () -> {
+                List<String> topicReplicas = KafkaTopicUtils.getKafkaTopicReplicasForEachPartition(
+                    namespaceName,
+                    topicName,
+                    scraperPodName,
+                    bootstrapServer
+                );
+                return topicReplicas.stream().anyMatch(line -> brokerIds.stream().anyMatch(line::contains));
+            }
+        );
+    }
+
     public static void waitForTopicWithPrefixDeletion(String namespaceName, String topicPrefix) {
         TestUtils.waitFor("deletion of all topics with prefix: " + topicPrefix, TestConstants.GLOBAL_POLL_INTERVAL, DELETION_TIMEOUT,
             () -> {

diff --git a/systemtest/src/test/java/io/strimzi/systemtest/cruisecontrol/CruiseControlST.java b/systemtest/src/test/java/io/strimzi/systemtest/cruisecontrol/CruiseControlST.java
@@ -771,10 +771,8 @@ void testAutoKafkaRebalanceScaleUpScaleDown() {
         // check that KafkaRebalance <cluster-name>-auto-rebalancing-<mode> is deleted
         KafkaRebalanceUtils.waitForKafkaRebalanceIsDeleted(testStorage.getNamespaceName(), KafkaResources.autoRebalancingKafkaRebalanceResourceName(testStorage.getClusterName(), KafkaAutoRebalanceMode.ADD_BROKERS));
 
-        LOGGER.info("Checking that Topic: {} has replicas on one of the new brokers (or both)", testStorage.getTopicName());
-        List<String> topicReplicas = KafkaTopicUtils.getKafkaTopicReplicasForEachPartition(testStorage.getNamespaceName(), testStorage.getTopicName(), scraperPodName, KafkaResources.plainBootstrapAddress(testStorage.getClusterName()));
-
-        assertTrue(topicReplicas.stream().anyMatch(line -> line.contains("3") || line.contains("4")));
+        KafkaTopicUtils.waitForTopicReplicasOnBrokers(testStorage.getNamespaceName(), testStorage.getTopicName(),
+            scraperPodName, KafkaResources.plainBootstrapAddress(testStorage.getClusterName()), Arrays.asList("3", "4"));
 
         LOGGER.info("Scaling Kafka down to {}", initialReplicas);