KAFKA-18105 Fix flaky PlaintextAdminIntegrationTest#testElectPreferredLeaders #20068

jim0987795064 · 2025-06-30T00:37:24Z

Changes

This PR improves the stability of the
PlaintextAdminIntegrationTest.testElectPreferredLeaders test by
introducing short Thread.sleep( ) delays before invoking:

changePreferredLeader( )
waitForBrokersOutOfIsr( )

Reasons

Metadata propagation for partition2 :
Kafka requires time to propagate the updated leader metadata across all
brokers. Without waiting, metadataCache may return outdated leader
information for partition2.
Eviction of broker1 from the ISR :
To simulate a scenario where broker1 is no longer eligible as leader,
the test relies on broker1 being removed from the ISR (e.g., due to
intentional shutdown). This eviction is not instantaneous and requires a
brief delay before Kafka reflects the change.

…r to improve test stability

chia7712 · 2025-06-30T15:18:53Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -2909,6 +2909,7 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
    TestUtils.assertLeader(client, partition2, 0)

    // Now change the preferred leader to 1
+    Thread.sleep(1000)


Could you please use "wait for condition" instead of time-based waiting?

…eader

frankvicky · 2025-07-05T14:49:12Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+        () => {
+          prior1 = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition1.topic, partition1.partition(), listenerName).get.id()
+          prior2 = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition2.topic, partition2.partition(), listenerName).get.id()
+          var m = Map.empty[TopicPartition, Optional[NewPartitionReassignment]]


Could you give this variable a meaningful name?

Hello @frankvicky,
I've already addressed this issue
Thanks for pointing it out!

chia7712 · 2025-07-05T19:28:57Z

core/src/test/scala/unit/kafka/utils/TestUtils.scala

@@ -1335,11 +1335,14 @@ object TestUtils extends Logging {
      case Failure(e) => throw e
    }

-    assertTrue(isLeaderElected, s"Timed out waiting for leader to become $expectedLeaderOpt. " +
+    assertTrue(isLeaderElected, s"Timed out waiting for leader to become expectedLeaderOpt. " +


Why do we need this change?

chia7712 · 2025-07-05T19:30:27Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

    // but shut it down...
    killBroker(1)
-    TestUtils.waitForBrokersOutOfIsr(client, Set(partition1, partition2), Set(1))
+    TestUtils.waitUntilTrue(


all we need is the waitForBrokersOutOfIsr, right?

chia7712 · 2025-07-05T19:33:50Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

      // Check the leader hasn't moved
-      TestUtils.assertLeader(client, partition1, prior1)
-      TestUtils.assertLeader(client, partition2, prior2)
+      TestUtils.waitUntilTrue(


It seems the way to stabilize the test is to ensure "all" brokers metadata get up-to-date, right?

m1a2st · 2025-07-07T02:02:56Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+    def sleepMillis(durationMs: Long): Unit = {
+      val startTime = System.currentTimeMillis()
+        TestUtils.waitUntilTrue(
+          () => System.currentTimeMillis() - startTime >= durationMs,
+          s"Waited less than $durationMs ms",
+          durationMs
+        )
+    }


Just out of curiosity, why do we need this method instead of using TimeUnit.MILLISECONDS.sleep()?

Hello @m1a2st , Thank you for pointing out this!
I've changed the first parameter of waitUntilTrue to check if metadata has been propagated to each broker

TaiJuWu · 2025-07-07T02:53:39Z

core/src/test/scala/unit/kafka/utils/TestUtils.scala

@@ -1298,11 +1298,11 @@ object TestUtils extends Logging {
    adminClient.incrementalAlterConfigs(configs)
  }

-  def assertLeader(client: Admin, topicPartition: TopicPartition, expectedLeader: Int): Unit = {


Do we need these changes in this file?

Hello @TaiJuWu, to use assertLeader as the first parameter of sleepMillisToPropagateMetadata, I think we need to change the return type of assertLeader.

Hello @TaiJuWu , I've fixed this issue

…l brokers.

…waitForLeaderToBecome to Unit

FrankYang0529 · 2025-07-09T02:41:29Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+    var prior2 = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition2.topic, partition2.partition(), listenerName).get.id()
+
+    def sleepMillisToPropagateMetadata(durationMs: Long): Unit = {
+      TimeUnit.MILLISECONDS.sleep(100)


In the following TestUtils.waitUntilTrue, it already do sleep if the condition is not matched. Do we need TimeUnit.MILLISECONDS.sleep(100) here?

Hello @FrankYang0529 , I think we can move TimeUnit.MILLISECONDS.sleep(100) here.
Thank you for pointing out!

TaiJuWu

Thanks for this patch, leave two nit.

TaiJuWu · 2025-07-09T03:54:52Z

core/src/test/scala/unit/kafka/utils/TestUtils.scala

@@ -1517,4 +1517,4 @@ object TestUtils extends Logging {
      timedOut.set(true)
    }
  }
-}
+}


Please revert this change.

TaiJuWu · 2025-07-09T03:55:09Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -4295,4 +4320,4 @@ object PlaintextAdminIntegrationTest {

    assertEquals(ServerLogConfigs.COMPRESSION_TYPE_DEFAULT, configs.get(brokerResource).get(ServerConfigs.COMPRESSION_TYPE_CONFIG).value)
  }
-}


Please revert this change.

Hello @TaiJuWu , I've addressed this issue.

…er function

m1a2st

Thanks for the update, left some comments

m1a2st · 2025-07-10T15:29:02Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+    TimeUnit.MILLISECONDS.sleep(1000)
+    val prior = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).get.id()


Could you add a comment explaining why we need to call sleep here?

How can we ensure that the first broker’s metadata is updated after 1000ms?

m1a2st · 2025-07-10T15:30:14Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

+          leaderId.contains(prior)
+        }
+        allSynced
+      }, s"Waited less than $durationMs ms", durationMs)


Could you update the error message to indicate which condition was not met?

Hello, I've added the comment and rewrite the error message in sleepMillisToPropagateMetadata.

m1a2st · 2025-07-10T15:31:31Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -2850,6 +2851,20 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
    }
  }

+  def sleepMillisToPropagateMetadata(durationMs: Long, partition: TopicPartition): Unit = {


Since durationMs is always set to 10000, we might consider removing this parameter.

Thanks for the suggestion.
I think it might be useful to keep durationMs as a parameter in case future changes require tuning the wait time.
This gives a bit more flexibility for potential modifications.

…agateMetadata

TaiJuWu · 2025-07-11T05:46:07Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -2850,6 +2851,21 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
    }
  }

+  def sleepMillisToPropagateMetadata(durationMs: Long, partition: TopicPartition): Unit = {


From my understanding, this function ensures that the leader update has propagated to all brokers’ metadata caches.
Is there a reason why this check isn't integrated directly into changePreferredLeader?
Please correct me if I'm wrong or miss something .

Hello @TaiJuWu ,I extracted this logic into a separate method so that it's easier to reuse in other places if needed. Keeping it outside makes the code more modular and improves readability.
Please let me know if you have any other questions.

Because there is no any real conditions to wait metadata propagation (you just wait a constant time), I think we can get real condition to wait.

For example, we can wait all brokers' partition leader is same.
We can do that by merging this function into changePreferredLeader .

chia7712 · 2025-07-12T14:41:26Z

core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala

@@ -2963,9 +2981,11 @@ class PlaintextAdminIntegrationTest extends BaseAdminIntegrationTest {
    TestUtils.assertLeader(client, partition2, 2)

    // Now change the preferred leader to 1
+    sleepMillisToPropagateMetadata(10000, partition2)


Perhaps it could be replaced by TestUtils.waitForPartitionMetadata(brokers, partition2.topic(), partition2.partition())?

Hello @chia7712 , after reviewing the waitForPartitionMetadata code, it seems that this API is used to check whether the partition leader broker is valid — that is, not -1 or unknown. It doesn't guarantee that all leader brokers are the same. When I run the code, the test still fails.

fix: Add sleep before changePreferredLeader and waitForBrokersOutOfIs…

dc2b489

…r to improve test stability

github-actions bot added triage PRs from the community core Kafka Broker tests Test fixes (including flaky tests) small Small PRs labels Jun 30, 2025

TaiJuWu added the ci-approved label Jun 30, 2025

chia7712 reviewed Jun 30, 2025

View reviewed changes

github-actions bot removed the triage PRs from the community label Jul 1, 2025

refactor: Replace Thread.sleep with waitUntilTrue in changePreferredL…

113cca0

…eader

frankvicky reviewed Jul 5, 2025

View reviewed changes

refactor: Rename variable of changePreferredLeader

94f10ca

chia7712 reviewed Jul 5, 2025

View reviewed changes

feat: Add TestUtils.waitUntilTrue-based sleepMillis helper

a0f2360

m1a2st reviewed Jul 7, 2025

View reviewed changes

TaiJuWu reviewed Jul 7, 2025

View reviewed changes

jim0987795064 added 8 commits July 7, 2025 14:27

revert: restore TestUtils.scala to original state

dfd30ba

revert: restore TestUtils.scala to original state

b2944b5

feat: Replace fixed sleep with a metadata propagation check across al…

4111838

…l brokers.

refactor: Move prior2 into waitUntilTrue

2f1ca60

revert: Recover the return type of assertLeader, assertNoLeader, and …

75c8d04

…waitForLeaderToBecome to Unit

refactor: Remove redundant code in waitForLeaderToBecome

1b09b82

refactor: Fix the indent

324ba84

refactor: Adjust the indent of sleepMillisToPropagateMetadata

335442c

FrankYang0529 reviewed Jul 9, 2025

View reviewed changes

TaiJuWu reviewed Jul 9, 2025

View reviewed changes

github-actions bot removed the small Small PRs label Jul 9, 2025

jim0987795064 force-pushed the KAFKA-18105 branch from 3537c84 to 335442c Compare July 9, 2025 07:17

github-actions bot added the small Small PRs label Jul 9, 2025

jim0987795064 force-pushed the KAFKA-18105 branch from ef22986 to 335442c Compare July 9, 2025 07:42

chore: revert the unnecessary change in TestUtils

f8e9870

refactor: Remove the unnecessary code in sleepMillisToPropagateMetadata

30e2350

github-actions bot removed the small Small PRs label Jul 9, 2025

jim0987795064 force-pushed the KAFKA-18105 branch 2 times, most recently from 8eead47 to 30e2350 Compare July 9, 2025 22:30

github-actions bot added the small Small PRs label Jul 9, 2025

refactor: Extract sleepMillisToPropagateMetadata as a standalone help…

47104c8

…er function

m1a2st reviewed Jul 10, 2025

View reviewed changes

chore: Add comment and Rewrite the error message in sleepMillisToprop…

bc8ca11

…agateMetadata

TaiJuWu reviewed Jul 11, 2025

View reviewed changes

chia7712 reviewed Jul 12, 2025

View reviewed changes

		TimeUnit.MILLISECONDS.sleep(1000)
		val prior = brokers.head.metadataCache.getPartitionLeaderEndpoint(partition.topic, partition.partition(), listenerName).get.id()

KAFKA-18105 Fix flaky PlaintextAdminIntegrationTest#testElectPreferredLeaders #20068

Are you sure you want to change the base?

KAFKA-18105 Fix flaky PlaintextAdminIntegrationTest#testElectPreferredLeaders #20068

Uh oh!

Conversation

jim0987795064 commented Jun 30, 2025 • edited by github-actions bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Changes

Reasons

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jim0987795064 Jul 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TaiJuWu left a comment

Choose a reason for hiding this comment

Uh oh!

TaiJuWu Jul 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

m1a2st left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

TaiJuWu Jul 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jim0987795064 commented Jun 30, 2025 •

edited by github-actions bot

Loading

jim0987795064 Jul 7, 2025 •

edited

Loading

TaiJuWu Jul 9, 2025 •

edited

Loading

TaiJuWu Jul 11, 2025 •

edited

Loading